/**
 * Copyright (c) 2021 OceanBase
 * OceanBase CE is licensed under Mulan PubL v2.
 * You can use this software according to the terms and conditions of the Mulan PubL v2.
 * You may obtain a copy of Mulan PubL v2 at:
 *          http://license.coscl.org.cn/MulanPubL-2.0
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PubL v2 for more details.
 */

#define USING_LOG_PREFIX SHARE
#include "ob_external_table_file_mgr.h"
#include "observer/dbms_scheduler/ob_dbms_sched_table_operator.h"
#include "share/catalog/ob_cached_catalog_meta_getter.h"
#include "share/external_table/ob_external_table_file_rpc_processor.h"
#include "share/external_table/ob_external_table_utils.h"
#include "share/schema/ob_iceberg_table_schema.h"
#include "sql/engine/cmd/ob_load_data_parser.h"
#include "sql/engine/table/ob_external_table_access_service.h"
#include "sql/resolver/ddl/ob_alter_table_resolver.h"
#include "sql/rewrite/ob_transform_utils.h"
#include "sql/table_format/iceberg/ob_iceberg_table_metadata.h"
#include "sql/table_format/iceberg/scan/task.h"
#include "storage/tablelock/ob_lock_inner_connection_util.h"
#include "sql/table_format/hive/ob_hive_table_metadata.h"
#include "share/catalog/hive/ob_hive_metastore.h"
#include "lib/file/ob_string_util.h"

namespace oceanbase
{
using namespace observer;
using namespace common;
using namespace sql;
using namespace transaction::tablelock;
using namespace pl;
using namespace common::sqlclient;
using namespace dbms_scheduler;
namespace share
{

int ObExternalFileInfo::deep_copy(ObIAllocator &allocator, const ObExternalFileInfo &other)
{
  int ret = OB_SUCCESS;
  if (OB_FAIL(ob_write_string(allocator, other.file_url_, this->file_url_))) {
    LOG_WARN("fail to write string", K(ret));
  } else if (OB_FAIL(ob_write_string(allocator, other.session_id_, this->session_id_))) {
    LOG_WARN("failed to write session id string", K(ret));
  } else if (OB_FAIL(ob_write_string(allocator, other.content_digest_, this->content_digest_))) {
    LOG_WARN("fail to write string", K(ret));
  } else {
    this->file_id_ = other.file_id_;
    this->part_id_ = other.part_id_;
    this->file_addr_ = other.file_addr_;
    this->file_size_ = other.file_size_;
    this->modify_time_ = other.modify_time_;
    this->row_start_ = other.row_start_;
    this->row_count_ = other.row_count_;

    this->pos_del_files_.reset();
    for (int64_t i = 0; OB_SUCC(ret) && i < other.pos_del_files_.count(); ++i) {
      ObString copied_str;
      if (OB_FAIL(ob_write_string(allocator, other.pos_del_files_.at(i), copied_str))) {
        LOG_WARN("fail to write string array element", K(ret), K(i));
      } else if (OB_FAIL(this->pos_del_files_.push_back(copied_str))) {
        LOG_WARN("fail to push back string array element", K(ret), K(i));
      }
    }
  }
  return ret;
}

int ObExternalFileInfo::assign(const ObExternalFileInfo &other)
{
  int ret = OB_SUCCESS;
  this->file_url_ = other.file_url_;
  this->file_id_ = other.file_id_;
  this->part_id_ = other.part_id_;
  this->file_addr_ = other.file_addr_;
  this->file_size_ = other.file_size_;
  this->row_start_ = other.row_start_;
  this->row_count_ = other.row_count_;
  this->session_id_ = other.session_id_;
  this->modify_time_ = other.modify_time_;
  this->content_digest_ = other.content_digest_;
  OZ (this->pos_del_files_.assign(other.pos_del_files_));
  return ret;
}

OB_SERIALIZE_MEMBER(ObExternalFileInfo, file_url_, file_id_, file_addr_, file_size_, part_id_,
                    row_start_, row_count_, session_id_, pos_del_files_, modify_time_,
                    content_digest_);

int ObExternalTableFilesKey::deep_copy(char *buf, const int64_t buf_len, ObIKVCacheKey *&key) const
{
   int ret = OB_SUCCESS;
   ObExternalTableFilesKey *new_value = NULL;
   ObDataBuffer allocator(buf, buf_len);
   if (OB_ISNULL(new_value = OB_NEWx(ObExternalTableFilesKey, &allocator))) {
     ret = OB_ALLOCATE_MEMORY_FAILED;
     LOG_WARN("fail to allocate memory", K(ret));
   } else {
     new_value->tenant_id_ = this->tenant_id_;
     new_value->table_id_ = this->table_id_;
     new_value->partition_id_ = this->partition_id_;
     key = new_value;
   }
   return ret;
 }

uint64_t ObExternalTableFileListKey::hash() const
{
  uint64_t hash_code = 0;
  hash_code = murmurhash(&tenant_id_, sizeof(uint64_t), hash_code);
  hash_code = path_.hash(hash_code);
  return hash_code;
}

int ObExternalTableFileListKey::deep_copy(char *buf, const int64_t buf_len, ObIKVCacheKey *&key) const
{
  int ret = OB_SUCCESS;
  ObExternalTableFileListKey *new_value = NULL;
  ObDataBuffer allocator(buf, buf_len);
  if (OB_ISNULL(new_value = OB_NEWx(ObExternalTableFileListKey, &allocator))) {
    ret = OB_ALLOCATE_MEMORY_FAILED;
    LOG_WARN("fail to allocate memory", K(ret));
  } else {
    new_value->tenant_id_ = this->tenant_id_;
    OZ(ob_write_string(allocator, this->path_, new_value->path_));
    OX(key = new_value);
  }
  return ret;
}

bool ObExternalTablePartitionsKey::operator==(const ObIKVCacheKey &other) const
{
  const ObExternalTablePartitionsKey &other_key
      = reinterpret_cast<const ObExternalTablePartitionsKey &>(other);
  bool r = this->tenant_id_ == other_key.tenant_id_
           && this->catalog_id_ == other_key.catalog_id_
           && this->db_name_.case_compare(other_key.db_name_) == 0
           && this->tb_name_.case_compare(other_key.tb_name_) == 0;
  return r;
}

uint64_t ObExternalTablePartitionsKey::hash() const
{
  uint64_t hash_code = 0;
  hash_code = murmurhash(&tenant_id_, sizeof(uint64_t), hash_code);
  hash_code = murmurhash(&catalog_id_, sizeof(uint64_t), hash_code);
  hash_code = db_name_.hash(hash_code);
  hash_code = tb_name_.hash(hash_code);
  return hash_code;
}

int ObExternalTablePartitionsKey::deep_copy(char *buf, const int64_t buf_len, ObIKVCacheKey *&key) const
{
  int ret = OB_SUCCESS;
  ObExternalTablePartitionsKey *new_value = NULL;
  ObDataBuffer allocator(buf, buf_len);
  if (OB_ISNULL(new_value = OB_NEWx(ObExternalTablePartitionsKey, &allocator))) {
    ret = OB_ALLOCATE_MEMORY_FAILED;
    LOG_WARN("fail to allocate memory", K(ret));
  } else {
    new_value->tenant_id_ = this->tenant_id_;
    new_value->catalog_id_ = this->catalog_id_;
    OZ(ob_write_string(allocator, this->db_name_, new_value->db_name_));
    OZ(ob_write_string(allocator, this->tb_name_, new_value->tb_name_));
    OX(key = new_value);
  }
  return ret;
}

int ObExternalTablePartitions::deep_copy(char *buf,
                                        const int64_t buf_len,
                                        ObIKVCacheValue *&value) const
{
  int ret = OB_SUCCESS;
  ObExternalTablePartitions *new_value = NULL;
  ObDataBuffer allocator(buf, buf_len);
  if (OB_ISNULL(new_value = OB_NEWx(ObExternalTablePartitions, &allocator))) {
    ret = OB_ALLOCATE_MEMORY_FAILED;
    LOG_WARN("fail to allocate memory", K(ret));
  } else {
    OX(new_value->create_ts_ = this->create_ts_);
    if (this->partition_infos_.count() > 0
        && OB_FAIL(new_value->partition_infos_.allocate_array(allocator,
                                                              this->partition_infos_.count()))) {
      LOG_WARN("Allocate PartitionInfo from array error", K(ret));
    }
    for (int64_t i = 0; OB_SUCC(ret) && i < this->partition_infos_.count(); ++i) {
      const PartitionInfo &old_partition_info = this->partition_infos_.at(i);
      PartitionInfo &new_partition_info = new_value->partition_infos_.at(i);
      OZ(ob_write_string(allocator,
                         old_partition_info.partition_,
                         new_partition_info.partition_));
      OZ(ob_write_string(allocator,
                         old_partition_info.path_,
                         new_partition_info.path_));
      OX(new_partition_info.modify_ts_ = old_partition_info.modify_ts_);

      const common::ObArrayWrap<ObString> &old_partition_values
          = this->partition_infos_.at(i).partition_values_;
      if (old_partition_values.count() > 0) {
        common::ObArrayWrap<ObString> &new_partition_values = new_partition_info.partition_values_;
        OZ(new_partition_values.allocate_array(allocator, old_partition_values.count()));
        for (int64_t j = 0; OB_SUCC(ret) && j < old_partition_values.count(); ++j) {
          OZ(ob_write_string(allocator, old_partition_values.at(j), new_partition_values.at(j)));
        }
      }
    }
    OX(value = new_value);
  }
  return ret;
}

int64_t ObExternalTableFiles::size() const
{
  int64_t size = sizeof(*this) + sizeof(ObString) * file_urls_.count()
                 + sizeof(ObString) * content_digests_.count() + sizeof(int64_t) * file_ids_.count()
                 + sizeof(int64_t) * file_sizes_.count()
                 + sizeof(int64_t) * modify_times_.count();
  for (int i = 0; i < file_urls_.count(); ++i) {
    size += file_urls_.at(i).length();
  }
  for (int i = 0; i < content_digests_.count(); ++i) {
    size += content_digests_.at(i).length();
  }
  return size;
 }

int ObExternalTableFiles::deep_copy(char *buf, const int64_t buf_len, ObIKVCacheValue *&value) const
{
  int ret = OB_SUCCESS;
  ObDataBuffer allocator(buf, buf_len);
  ObExternalTableFiles *new_value = NULL;
  if (OB_ISNULL(new_value = OB_NEWx(ObExternalTableFiles, &allocator))) {
    ret = OB_ALLOCATE_MEMORY_FAILED;
    LOG_WARN("fail to allocate memory", K(ret));
  }

  if (OB_SUCC(ret) && this->file_urls_.count() > 0) {
    if (OB_FAIL(new_value->file_urls_.allocate_array(allocator, this->file_urls_.count()))) {
      LOG_WARN("fail to allocate array", K(ret));
    } else {
      for (int i = 0; OB_SUCC(ret) && i < this->file_urls_.count(); i++) {
        OZ (ob_write_string(allocator, this->file_urls_.at(i), new_value->file_urls_.at(i)));
      }
    }
  }

  if (OB_SUCC(ret) && this->content_digests_.count() > 0) {
    if (OB_FAIL(
          new_value->content_digests_.allocate_array(allocator, this->content_digests_.count()))) {
      LOG_WARN("fail to allocate array", K(ret));
    } else {
      for (int i = 0; OB_SUCC(ret) && i < this->content_digests_.count(); i++) {
        OZ(ob_write_string(allocator, this->content_digests_.at(i),
                           new_value->content_digests_.at(i)));
      }
    }
  }

  if (OB_SUCC(ret) && this->file_ids_.count() > 0) {
    if (OB_FAIL(new_value->file_ids_.allocate_array(allocator, this->file_ids_.count()))) {
      LOG_WARN("fail to allocate array", K(ret));
    } else {
      MEMCPY(new_value->file_ids_.get_data(), this->file_ids_.get_data(),
             sizeof(int64_t) * this->file_ids_.count());
    }
  }

  if (OB_SUCC(ret) && this->file_sizes_.count() > 0) {
    if (OB_FAIL(new_value->file_sizes_.allocate_array(allocator, this->file_sizes_.count()))) {
      LOG_WARN("fail to allocate array", K(ret));
    } else {
      MEMCPY(new_value->file_sizes_.get_data(), this->file_sizes_.get_data(),
             sizeof(int64_t) * this->file_sizes_.count());
    }
  }
  if (OB_SUCC(ret) && this->modify_times_.count() > 0) {
    if (OB_FAIL(new_value->modify_times_.allocate_array(allocator, this->modify_times_.count()))) {
      LOG_WARN("fail to allocate array", K(ret));
    } else {
      MEMCPY(new_value->modify_times_.get_data(), this->modify_times_.get_data(),
             sizeof(int64_t) * this->modify_times_.count());
    }
  }
  if (OB_SUCC(ret)) {
    new_value->create_ts_ = this->create_ts_;
    new_value->modify_ts_ = this->modify_ts_;
  }
  value = new_value;
  return ret;
}

int ObExternalTableFileManager::flush_cache(const uint64_t tenant_id, const uint64_t table_id, const uint64_t part_id)
{
  int ret = OB_SUCCESS;
  ObExternalTableFilesKey key;
  key.tenant_id_ = tenant_id;
  key.table_id_ = table_id;
  key.partition_id_ = part_id;
  if (OB_FAIL(kv_cache_.erase(key))) {
    if (OB_ENTRY_NOT_EXIST != ret) {
      LOG_WARN("fail to erase value", K(ret), K(key));
    } else {
      ret = OB_SUCCESS;
    }
  }
  return ret;
}

int ObExternalTableFileManager::clear_inner_table_files(
    const uint64_t tenant_id,
    const uint64_t table_id,
    ObMySQLTransaction &trans)
{
  int ret = OB_SUCCESS;
  ObSqlString delete_sql;
  int64_t affected_rows = 0;
  OZ (delete_sql.assign_fmt("DELETE FROM %s WHERE TABLE_ID = %lu",
                            OB_ALL_EXTERNAL_TABLE_FILE_TNAME, table_id));
  OZ (trans.write(tenant_id, delete_sql.ptr(), affected_rows));
  LOG_DEBUG("check clear rows", K(affected_rows));
  return ret;
}


int ObExternalTableFileManager::clear_inner_table_files_within_one_part(
    const uint64_t tenant_id,
    const uint64_t table_id,
    const uint64_t part_id,
    ObMySQLTransaction &trans)
{
  int ret = OB_SUCCESS;
  ObSqlString delete_sql;
  int64_t affected_rows = 0;
  OZ (delete_sql.assign_fmt("DELETE FROM %s WHERE TABLE_ID = %lu AND PART_ID = %lu",
                            OB_ALL_EXTERNAL_TABLE_FILE_TNAME, table_id, part_id));
  OZ (trans.write(tenant_id, delete_sql.ptr(), affected_rows));
  LOG_DEBUG("check clear rows", K(affected_rows));
  return ret;
}
int ObExternalTableFileManager::init()
{
  int ret = OB_SUCCESS;
  OZ (kv_cache_.init("external_table_file_cache"));
  OZ (external_file_list_cache_.init("external_table_file_list_cache"));
  OZ (external_partitions_cache_.init("external_table_partitions_cache"));
  return ret;
}

ObExternalTableFileManager &ObExternalTableFileManager::get_instance()
{
  static ObExternalTableFileManager instance_;
  return instance_;
}

int ObExternalTableFileManager::get_external_files(
    const uint64_t tenant_id,
    const uint64_t table_id,
    const bool is_local_file_on_disk,
    ObIAllocator &allocator,
    ObIArray<ObExternalFileInfo> &external_files,
    ObIArray<ObNewRange *> *range_filter /*default = NULL*/)
{
  return get_external_files_by_part_id(tenant_id, table_id, table_id, is_local_file_on_disk,  allocator, external_files, range_filter);
}

int ObExternalTableFileManager::get_external_files_by_part_ids(
    const uint64_t tenant_id,
    const uint64_t table_id,
    ObIArray<int64_t> &partition_ids,
    const bool is_local_file_on_disk,
    ObIAllocator &allocator,
    ObIArray<ObExternalFileInfo> &external_files,
    ObIArray<ObNewRange *> *range_filter /*default = NULL*/)
{
  int ret = OB_SUCCESS;
  if (partition_ids.empty()) {
     OZ (get_external_files_by_part_id(tenant_id, table_id, table_id, is_local_file_on_disk,  allocator, external_files, range_filter));
  } else {
    for (int64_t i = 0; OB_SUCC(ret) && i < partition_ids.count(); i++) {
      OZ (get_external_files_by_part_id(tenant_id, table_id, partition_ids.at(i), is_local_file_on_disk,  allocator, external_files, range_filter));
    }
  }
  return ret;
}

int ObExternalTableFileManager::get_mocked_external_table_files(
    const uint64_t tenant_id,
    ObIArray<int64_t> &partition_ids,
    sql::ObExecContext &ctx,
    const ObDASScanCtDef &das_ctdef,
    ObIArray<ObExternalFileInfo> &external_files)
{
  int ret = OB_SUCCESS;
  ObExprRegexpSessionVariables regexp_vars;
  ObSqlString full_path;
  ObArray<share::ObExternalTableBasicFileInfo> basic_file_infos;
  bool is_part_table = das_ctdef.table_param_.is_partition_table();

  ObExternalFileFormat::FormatType external_table_type;
  if (OB_FAIL(ObSQLUtils::get_external_table_type(das_ctdef.external_file_format_str_.str_,
                                                  external_table_type))) {
    LOG_WARN("failed to get external table type", K(ret));
  }

  if (OB_FAIL(ret)) {
  } else if (ObLakeTableFormat::HIVE == das_ctdef.lake_table_format_) {
    // HMS format does not need regexp_vars.
    bool is_local_file_on_disk
        = ObSQLUtils::is_external_files_on_local_disk(das_ctdef.external_file_location_.str_);
    if (OB_UNLIKELY(is_local_file_on_disk)) {
      ret = OB_ERR_UNEXPECTED;
      LOG_WARN("hms format files should not be on local disk", K(ret));
    } else if (OB_LIKELY(is_part_table)) {
      int64_t part_id = -1;
      bool found = false;
      ObString original_location;
      OZ(ob_write_string(ctx.get_allocator(),
                         das_ctdef.external_file_location_.str_,
                         original_location,
                         true /*c_style*/));
      share::ObExternalTablePartInfo part_info;
      int64_t file_id = 0;

      for (int64_t index = 0; OB_SUCC(ret) && index < partition_ids.count(); index++) {
        part_id = partition_ids.at(index);
        found = false;

        basic_file_infos.reuse();

        for (int64_t part_idx = 0;
             OB_LIKELY(!found) && part_idx < das_ctdef.partition_infos_.count();
             part_idx++) {
          part_info = das_ctdef.partition_infos_.at(part_idx);
          if (OB_LIKELY(part_info.part_id_ == part_id)) {
            found = true;
            ObString partition_spec;
            OZ(ob_write_string(ctx.get_allocator(),
                               part_info.partition_spec_,
                               partition_spec,
                               true /*c_style*/));

            // Handle related partition files
            // Partition spec is same as the target location path.
            // Such as year=2025/month=1/day=1 to
            // hdfs://localhost/user/hive/warehouse/test_table/year=2025/month=1/day=1/xxx.
            // Our target is only collecting the related partition files not all useless files of
            // partition.
            ObSqlString target_location;
            if (partition_spec.find(':') != nullptr) {
              OZ(target_location.assign(partition_spec));
            } else {
              OZ(target_location.append_fmt("%s/%s", original_location.ptr(), partition_spec.ptr()));
            }
            // Get all related partition files
            // Note: full path should be partition spec str.
            ObSqlString fake_full_path;
            OZ(fake_full_path.append(partition_spec));
            OZ(ObExternalTableUtils::collect_external_file_list(
                ctx.get_my_session(),
                tenant_id,
                das_ctdef.ref_table_id_,
                target_location.string(),
                das_ctdef.external_file_access_info_.str_,
                das_ctdef.external_file_pattern_.str_,
                common::ObString::make_empty_string(),
                is_part_table,
                regexp_vars,
                ctx.get_allocator(),
                fake_full_path,
                basic_file_infos));
            for (int64_t k = 0; OB_SUCC(ret) && k < basic_file_infos.count(); k++) {
              ObExternalFileInfo file_info;
              file_info.file_id_ = ++file_id;
              file_info.file_size_ = basic_file_infos.at(k).size_;
              file_info.part_id_ = part_id;
              file_info.file_url_ = basic_file_infos.at(k).url_;
              file_info.modify_time_ = basic_file_infos.at(k).last_modify_time_;
              file_info.content_digest_ = basic_file_infos.at(k).content_digest_;
              OZ(external_files.push_back(file_info));
              LOG_TRACE("get file detail", K(ret), K(part_id), K(part_info), K(file_info));
            }
          }
        }
      }
    } else {
      OZ(ObExternalTableUtils::collect_external_file_list(ctx.get_my_session(),
                                                          tenant_id,
                                                          das_ctdef.ref_table_id_,
                                                          das_ctdef.external_file_location_.str_,
                                                          das_ctdef.external_file_access_info_.str_,
                                                          das_ctdef.external_file_pattern_.str_,
                                                          common::ObString::make_empty_string(),
                                                          is_part_table,
                                                          regexp_vars,
                                                          ctx.get_allocator(),
                                                          full_path,
                                                          basic_file_infos));
      for (int i = 0; OB_SUCC(ret) && i < basic_file_infos.count(); i++) {
        ObExternalFileInfo file;
        file.file_id_ = i + 1;
        file.file_size_ = basic_file_infos.at(i).size_;
        ObString file_url = basic_file_infos.at(i).url_;
        file.file_url_ = file_url;
        file.part_id_ = 0;
        file.modify_time_ = basic_file_infos.at(i).last_modify_time_;
        file.content_digest_ = basic_file_infos.at(i).content_digest_;
        OZ(external_files.push_back(file));
      }
    }
  } else if (ObExternalFileFormat::FormatType::ODPS_FORMAT == external_table_type) {
    if (is_part_table) {
      for (int64_t i = 0; OB_SUCC(ret) && i < partition_ids.count(); i++) {
        int64_t part_id = partition_ids.at(i);
        if (part_id != INT64_MAX) {
          ObExternalFileInfo file;
          bool found = false;
          for (int64_t j = 0; OB_SUCC(ret) && !found && j < das_ctdef.partition_infos_.count(); j++) {
            share::ObExternalTablePartInfo part_info = das_ctdef.partition_infos_.at(j);
            if (part_id == part_info.part_id_) {
              file.file_url_ = part_info.partition_spec_;
              file.part_id_ = part_id;
              found = true;
            }
          }

          if (found) {
            file.file_size_ = -1;
            file.file_id_ = i + 1;
            file.file_addr_ = GCTX.self_addr();
            OZ(external_files.push_back(file));
          }
        }
      }
    } else {
      ObExternalFileInfo file;
      file.file_size_ = -1;
      file.file_id_ = 1;
      file.file_addr_ = GCTX.self_addr();
      OZ(external_files.push_back(file));
    }
  } else if (ObExternalFileFormat::PARQUET_FORMAT == external_table_type
             || ObExternalFileFormat::ORC_FORMAT == external_table_type
             || ObExternalFileFormat::CSV_FORMAT == external_table_type) {
    // Means orc, parquet, csv format
    OZ(ctx.get_my_session()->get_regexp_session_vars(regexp_vars));
    OZ(ObExternalTableUtils::collect_external_file_list(ctx.get_my_session(),
                                                        tenant_id,
                                                        das_ctdef.ref_table_id_,
                                                        das_ctdef.external_file_location_.str_,
                                                        das_ctdef.external_file_access_info_.str_,
                                                        das_ctdef.external_file_pattern_.str_,
                                                        common::ObString::make_empty_string(),
                                                        is_part_table,
                                                        regexp_vars,
                                                        ctx.get_allocator(),
                                                        full_path,
                                                        basic_file_infos));
    bool is_local_file_on_disk
        = ObSQLUtils::is_external_files_on_local_disk(das_ctdef.external_file_location_.str_);
    for (int i = 0; OB_SUCC(ret) && i < basic_file_infos.count(); i++) {
      ObExternalFileInfo file;
      file.file_id_ = i + 1;
      file.file_size_ = basic_file_infos.at(i).size_;
      ObString file_url = basic_file_infos.at(i).url_;
      if (is_local_file_on_disk) {
        ObString ip_port = file_url.split_on(ip_delimiter);
        OZ (file.file_addr_.parse_from_string(ip_port));
      }
      file.file_url_ = file_url;
      file.part_id_ = is_part_table ? i + 1 : 0;
      file.modify_time_ = basic_file_infos.at(i).last_modify_time_;
      file.content_digest_ = basic_file_infos.at(i).content_digest_;
      OZ (external_files.push_back(file));
    }
  } else {
    ret = OB_NOT_SUPPORTED;
    LOG_WARN("unsupported table format", K(ret), K(external_table_type));
  }
  return ret;
}

int ObExternalTableFileManager::get_external_files_by_part_id(
    const uint64_t tenant_id,
    const uint64_t table_id,
    const uint64_t partition_id,
    const bool is_local_file_on_disk,
    ObIAllocator &allocator,
    ObIArray<ObExternalFileInfo> &external_files,
    ObIArray<ObNewRange *> *range_filter /*default = NULL*/)
{
  int ret = OB_SUCCESS;
  ObKVCacheHandle handle;
  const ObExternalTableFiles *ext_files = NULL;
  ObExternalTableFilesKey key;
  key.tenant_id_ = tenant_id;
  key.table_id_ = table_id;
  key.partition_id_ = partition_id;
  if (OB_FAIL(kv_cache_.get(key, ext_files, handle))) {
    if (OB_ENTRY_NOT_EXIST != ret) {
      LOG_WARN("fail to get from KVCache", K(ret), K(key));
    }
  }

  if ((OB_SUCC(ret) && is_cache_value_timeout(*ext_files))
      || OB_ENTRY_NOT_EXIST == ret) {
    if (OB_FAIL(fill_cache_from_inner_table(key, ext_files, handle))) {
      LOG_WARN("fail to fill cache from inner table", K(ret));
    }
  }
  for (int i = 0; OB_SUCC(ret) && i < ext_files->file_urls_.count(); ++i) {
    bool in_ranges = false;
    if (range_filter != NULL && OB_FAIL(ObExternalTableUtils::is_file_id_in_ranges(*range_filter,
                                                                        ext_files->file_ids_.at(i),
                                                                        in_ranges))) {
      LOG_WARN("failed to judge file id in ranges", K(ret));
    } else if (range_filter == NULL || in_ranges) {
      ObExternalFileInfo file_info;
      ObString file_url = ext_files->file_urls_.at(i);
      file_info.file_id_ = ext_files->file_ids_.at(i);
      file_info.file_size_ = ext_files->file_sizes_.at(i);
      file_info.part_id_ = partition_id;
      file_info.modify_time_ = ext_files->modify_times_.at(i);
      ObString content_digest = ext_files->content_digests_.at(i);
      if (is_local_file_on_disk) {
        ObString ip_port = file_url.split_on(ip_delimiter);
        OZ (file_info.file_addr_.parse_from_string(ip_port));
      }
      OZ (ob_write_string(allocator, file_url, file_info.file_url_));
      OZ (ob_write_string(allocator, content_digest, file_info.content_digest_));
      OZ (external_files.push_back(file_info));
    }
  }
  LOG_TRACE("get external file list result", K(table_id), K(is_local_file_on_disk), K(external_files));
  return ret;
}


int ObExternalTableFileManager::build_row_for_file_name(ObNewRow &row, ObIAllocator &allocator)
{
  int ret = OB_SUCCESS;
  ObObj *obj_array = nullptr;
  const int size = 1;
  if (OB_ISNULL(obj_array = static_cast<ObObj*>(
              allocator.alloc(sizeof(ObObj) * size)))) {
    ret = OB_ALLOCATE_MEMORY_FAILED;
    LOG_WARN("fail to allocate memory", K(ret));
  } else {
    for (ObObj *ptr = obj_array; ptr < obj_array + size; ++ptr) {
      new(ptr)ObObj();
      ptr->set_collation_type(ObCharset::get_system_collation());
    }
    row.assign(obj_array, size);
  }
  return ret;
}

int ObExternalTableFileManager::get_genarated_expr_from_partition_column(const ObColumnSchemaV2 *column_schema,
                                                                         const ObTableSchema *table_schema,
                                                                         ObSQLSessionInfo *session_info,
                                                                         ObRawExprFactory *expr_factory,
                                                                         ObSchemaGetterGuard &schema_guard,
                                                                         ObIAllocator &allocator,
                                                                         ObRawExpr *&gen_expr)
{
  int ret = OB_SUCCESS;
  if (OB_ISNULL(column_schema)) {
    ret = OB_ERR_UNEXPECTED;
    LOG_WARN("unexpected error", K(ret));
  } else if (column_schema->is_generated_column()) {
    ObColumnSchemaV2 tmp;
    ObSchemaChecker schema_checker;
    ObResolverParams params;
    params.session_info_ = session_info;
    params.allocator_ = &allocator;
    params.expr_factory_ = expr_factory;
    ObString col_def;
    if (OB_FAIL(schema_checker.init(schema_guard))) {
      LOG_WARN("init schema checker failed", K(ret));
    } else if (FALSE_IT(params.schema_checker_ = &schema_checker)) {
    } else if (OB_FAIL(column_schema->get_orig_default_value().get_string(col_def))) {
      LOG_WARN("get generated column definition failed", K(ret));
    } else if (OB_FAIL(ObResolverUtils::resolve_generated_column_expr(params, col_def, const_cast<ObTableSchema &>(*table_schema),
                                                                      tmp, gen_expr))) {
      LOG_WARN("resolve generated column expr failed", K(ret));
    } else if (OB_FAIL(gen_expr->formalize(session_info))) {
      LOG_WARN("formalize failed", K(ret));
    } else {
      ObRawExprResType expected_type;
      expected_type.set_meta(column_schema->get_meta_type());
      expected_type.set_accuracy(column_schema->get_accuracy());
      expected_type.set_result_flag(ObRawExprUtils::calc_column_result_flag(*column_schema));
      if (ObRawExprUtils::need_column_conv(expected_type, *gen_expr, true)) {
        if (OB_FAIL(ObRawExprUtils::build_column_conv_expr(*expr_factory, column_schema, gen_expr, session_info))) {
          LOG_WARN("create cast expr failed", K(ret));
        }
      }
    }
  } else {
    ret = OB_ERR_UNEXPECTED;
    LOG_WARN("external table partition expr is not generated column", K(ret));
  }
  return ret;
}

int ObExternalTableFileManager::cg_expr_by_mocking_field_expr(const ObTableSchema *table_schema,
                                                              ObRawExpr *gen_expr,
                                                              ObSQLSessionInfo *session_info,
                                                              ObRawExprFactory *expr_factory,
                                                              share::schema::ObSchemaGetterGuard &schema_guard,
                                                              ObIAllocator &allocator,
                                                              ObTempExpr *&temp_expr)
{
  int ret = OB_SUCCESS;
  ObColumnRefRawExpr *mock_field_expr = NULL;
  RowDesc row_desc;
  ObRawExpr *file_name_expr = NULL;
  if (OB_FAIL(expr_factory->create_raw_expr(T_REF_COLUMN, mock_field_expr))) {
    LOG_WARN("create column ref raw expr failed", K(ret));
  } else if (OB_ISNULL(mock_field_expr)) {
    ret = OB_ERR_UNEXPECTED;
  } else if (OB_FALSE_IT(mock_field_expr->set_column_attr(table_schema->get_table_name(), "_file_name"))) {
    LOG_WARN(("field_expr is null"));
  } else if (OB_FAIL(mock_field_expr->add_flag(IS_COLUMN))) {
    LOG_WARN("failed to add flag IS_COLUMN", K(ret));
  } else if (OB_FAIL(row_desc.init())) {
    LOG_WARN("row desc init failed", K(ret));
  } else if (OB_FAIL(row_desc.add_column(mock_field_expr))) {
    LOG_WARN("fail to add column", K(ret));
  } else if (OB_FAIL(ObRawExprUtils::extract_metadata_filename_expr(gen_expr, file_name_expr))) {
    LOG_WARN("get metadata filename column failed", K(ret));
  } else if (OB_ISNULL(file_name_expr)) {
    ret = OB_ERR_UNEXPECTED;
    LOG_WARN("gen part expr of ext table with no metadata file_name", K(ret));
  } else if (OB_FALSE_IT(mock_field_expr->set_collation_type(file_name_expr->get_result_meta().get_collation_type()))) {
  } else if (OB_FALSE_IT(mock_field_expr->set_data_type(file_name_expr->get_result_meta().get_type()))) {
  } else if (OB_FAIL(ObTransformUtils::replace_expr(file_name_expr,
                                                    mock_field_expr,
                                                    gen_expr))) {
    LOG_WARN("replace exprs failed", K(ret));
  } else if (OB_FAIL(ObStaticEngineExprCG::gen_expr_with_row_desc(gen_expr,
                                                                      row_desc,
                                                                      allocator,
                                                                      session_info,
                                                                      &schema_guard,
                                                                      temp_expr))) {
    LOG_WARN("fail to gen temp expr", K(ret));
  }
  return ret;
}

int ObExternalTableFileManager::cg_partition_expr_rt_expr(const ObTableSchema *table_schema,
                                                          ObRawExprFactory *expr_factory,
                                                          ObSQLSessionInfo *session_info,
                                                          share::schema::ObSchemaGetterGuard &schema_guard,
                                                          ObIAllocator &allocator,
                                                          ObIArray<ObTempExpr *> &temp_exprs)
{
  int ret = OB_SUCCESS;
  ObSEArray<uint64_t, 4> partition_column_ids;
  if (OB_ISNULL(expr_factory) || OB_ISNULL(table_schema) || OB_ISNULL(session_info)) {
    ret = OB_ERR_UNEXPECTED;
    LOG_WARN("unexpected null", K(ret), K(expr_factory), K(table_schema), K(session_info));
  }
  OZ (table_schema->get_partition_key_info().get_column_ids(partition_column_ids));
  for (int64_t i = 0; OB_SUCC(ret) && i < partition_column_ids.count(); i++) {
    const ObColumnSchemaV2 *column_schema = table_schema->get_column_schema(partition_column_ids.at(i));
    ObString expr_str = column_schema->get_cur_default_value().get_string();
    if (column_schema->is_generated_column()) {
      ObRawExpr *gen_expr = NULL;
      ObTempExpr *temp_expr = NULL;
      if (OB_FAIL(get_genarated_expr_from_partition_column(column_schema, table_schema, session_info,
                                                           expr_factory, schema_guard, allocator, gen_expr))) {
        LOG_WARN("get generated expr from partition column failed", K(ret));
      } else if (OB_FAIL(cg_expr_by_mocking_field_expr(table_schema, gen_expr, session_info,
                                                       expr_factory, schema_guard, allocator, temp_expr))) {
        LOG_WARN("cg expr failed", K(ret));
      } else if (OB_FAIL(temp_exprs.push_back(temp_expr))) {
        LOG_WARN("push back failed", K(ret));
      }
    } else {
      ret = OB_ERR_UNEXPECTED;
      LOG_WARN("external table partition expr is not generated column", K(ret));
    }
  }
  return ret;
}

int ObExternalTableFileManager::find_partition_existed(ObIArray<ObNewRow> &existed_part,
                                                       ObNewRow &file_part_val,
                                                       int64_t &found)
{
  int ret = OB_SUCCESS;
  found = -1;
  for (int64_t i = 0; OB_SUCC(ret) && found == -1 && i < existed_part.count(); i++) {
    bool match = true;
    CK (OB_LIKELY(existed_part.at(i).get_count() == file_part_val.get_count()));
    for (int64_t j = 0; OB_SUCC(ret) && match && j < file_part_val.get_count(); j++) {
      if (!existed_part.at(i).get_cell(j).can_compare(file_part_val.get_cell(j))) {
        ret = OB_ERR_UNEXPECTED;
        LOG_WARN("can not compare obj", K(existed_part.at(i)), K(file_part_val.get_cell(j)));
      } else if (existed_part.at(i).get_cell(j) == file_part_val.get_cell(j)) {
        //do nothing
      } else {
        match = false;
      }
    }
    if (match) {
      found = i;
    }
  }
  return ret;
}

int ObExternalTableFileManager::alter_partition_for_ext_table(ObMySQLTransaction &trans,
                                                               sql::ObExecContext &exec_ctx,
                                                               ObAlterTableStmt *alter_table_stmt,
                                                               ObIArray<int64_t> &file_part_ids)
{
  int ret = OB_SUCCESS;
  UNUSED(trans); //TODO: use the same trans to create partition
  obrpc::ObCommonRpcProxy *common_rpc_proxy = NULL;
  ObTaskExecutorCtx *task_exec_ctx = NULL;
  ObAlterTableRes res;
  if (OB_FAIL(ret)) {
  } else if (OB_ISNULL(task_exec_ctx = GET_TASK_EXECUTOR_CTX(exec_ctx))) {
    ret = OB_NOT_INIT;
    LOG_WARN("get task executor context failed");
  } else if (OB_FAIL(task_exec_ctx->get_common_rpc(common_rpc_proxy))) {
    LOG_WARN("get common rpc proxy failed", K(ret));
  } else if (OB_ISNULL(common_rpc_proxy)){
    ret = OB_ERR_UNEXPECTED;
    LOG_WARN("common rpc proxy should not be null", K(ret));
  } else if (OB_FAIL(common_rpc_proxy->alter_table(alter_table_stmt->get_alter_table_arg(), res))) {
    LOG_WARN("alter table failed", K(ret));
  } else if (OB_UNLIKELY(res.res_arg_array_.count() <= 0)) {
    ret = OB_ERR_UNEXPECTED;
    LOG_WARN("create partition may be failed", K(ret));
  } else {
    for (int64_t i = 0; OB_SUCC(ret) && i < res.res_arg_array_.count(); i++) {
      OZ (file_part_ids.push_back(res.res_arg_array_.at(i).part_object_id_));
    }
  }
  return ret;
}

int ObExternalTableFileManager::add_partition_for_alter_stmt(ObAlterTableStmt *&alter_table_stmt,
                                                             const ObString &part_name,
                                                             ObNewRow &part_val)
{
  int ret = OB_SUCCESS;
  ObPartition partition;
  if (OB_ISNULL(alter_table_stmt)) {
    ret = OB_ERR_UNEXPECTED;
    LOG_WARN("add partition for alter stmt failed", K(ret));
  } else if (OB_FAIL(partition.set_part_name(const_cast<ObString &>(part_name)))) {
    LOG_WARN("set partition name failed", K(ret));
  } else if (OB_FAIL(partition.add_list_row(part_val))) {
    LOG_WARN("add list row failed", K(ret));
  } else if (OB_FAIL(alter_table_stmt->get_alter_table_arg().alter_table_schema_.add_partition(partition))) {
    LOG_WARN("failed to add partition", K(ret));
  }
  return ret;
}

int ObExternalTableFileManager::create_alter_table_stmt(sql::ObExecContext &exec_ctx,
                                                        const ObTableSchema *table_schema,
                                                        const ObDatabaseSchema *database_schema,
                                                        const int64_t part_num,
                                                        const ObAlterTableArg::AlterPartitionType alter_part_type,
                                                        ObAlterTableStmt *&alter_table_stmt)
{
  int ret = OB_SUCCESS;
  ObSQLSessionInfo *session_info = exec_ctx.get_my_session();
  ObResolverParams params;
  CK (OB_NOT_NULL(session_info));
  CK (OB_NOT_NULL(table_schema));
  CK (OB_NOT_NULL(database_schema));
  params.expr_factory_ = exec_ctx.get_expr_factory();
  params.stmt_factory_ = exec_ctx.get_stmt_factory();
  params.allocator_ = &exec_ctx.get_allocator();
  params.session_info_ = exec_ctx.get_my_session();
  params.query_ctx_ = exec_ctx.get_stmt_factory()->get_query_ctx();
  SMART_VAR (ObAlterTableResolver, alter_table_resolver, params) {
    if (alter_table_stmt == NULL &&
        NULL == (alter_table_stmt = alter_table_resolver.create_stmt<ObAlterTableStmt>())) {
      ret = OB_ALLOCATE_MEMORY_FAILED;
      SQL_RESV_LOG(ERROR, "failed to create alter table stmt", K(ret));
    } else if (OB_FAIL(alter_table_stmt->set_tz_info_wrap(session_info->get_tz_info_wrap()))) {
      SQL_RESV_LOG(WARN, "failed to set_tz_info_wrap", "tz_info_wrap", session_info->get_tz_info_wrap(), K(ret));
    } else if (OB_FAIL(alter_table_stmt->set_nls_formats(
        session_info->get_local_nls_date_format(),
        session_info->get_local_nls_timestamp_format(),
        session_info->get_local_nls_timestamp_tz_format()))) {
      SQL_RESV_LOG(WARN, "failed to set_nls_formats", K(ret));
    } else if (OB_FAIL(alter_table_stmt->set_origin_database_name(database_schema->get_database_name_str()))) {
      SQL_RESV_LOG(WARN, "failed to set origin database name", K(ret));
    } else if (OB_FAIL(alter_table_stmt->set_origin_table_name(table_schema->get_table_name_str()))) {
      SQL_RESV_LOG(WARN, "failed to set origin table name", K(ret));
    } else {
      alter_table_stmt->set_tenant_id(table_schema->get_tenant_id());
      alter_table_stmt->set_table_id(table_schema->get_table_id());
      alter_table_stmt->get_alter_table_arg().alter_table_schema_.set_schema_version(
                                                                                table_schema->get_schema_version());
      alter_table_stmt->get_alter_table_arg().alter_table_schema_.
                          get_part_option().set_part_func_type(table_schema->get_part_option().get_part_func_type());
      alter_table_stmt->get_alter_table_arg().alter_table_schema_.set_part_num(part_num);
      alter_table_stmt->set_alter_table_partition();
      alter_table_stmt->get_alter_table_arg().alter_part_type_ = alter_part_type;
      alter_table_stmt->get_alter_table_arg().alter_table_schema_.alter_type_ = OB_DDL_ALTER_TABLE;
      alter_table_stmt->get_alter_table_arg().alter_table_schema_.set_part_level(table_schema->get_part_level());
      alter_table_stmt->get_alter_table_arg().tz_info_ = session_info->get_tz_info_wrap().get_tz_info_offset();
      alter_table_stmt->get_alter_table_arg().is_inner_ = session_info->is_inner();
      alter_table_stmt->get_alter_table_arg().exec_tenant_id_ = session_info->get_effective_tenant_id();
      alter_table_stmt->get_alter_table_arg().session_id_ = session_info->get_sessid_for_table();
      alter_table_stmt->get_alter_table_arg().alter_table_schema_.set_table_flags(table_schema->get_table_flags());
    }
  }
  return ret;
}

int ObExternalTableFileManager::get_all_partition_list_val(const ObTableSchema *table_schema, ObIArray<ObNewRow> &part_vals, ObIArray<int64_t> &part_ids)
{
  int ret = OB_SUCCESS;
  CK (OB_NOT_NULL(table_schema));
  if (table_schema->get_partition_num() > 0) {
    CK(OB_NOT_NULL(table_schema->get_part_array()));
  }
  for (int64_t i = 0; OB_SUCC(ret) && i < table_schema->get_partition_num(); i++) {
    ObPartition *partition = table_schema->get_part_array()[i];
    if (OB_ISNULL(partition)) {
      ret = OB_ERR_UNEXPECTED;
      LOG_WARN("partition is null", K(ret));
    } else if (partition->get_part_name().compare("P_DEFAULT") == 0) {
      //skip
    } else {
      const ObIArray<ObNewRow> &list_val = partition->get_list_row_values();
      if (OB_LIKELY(list_val.count() == 1)) {
        const ObNewRow &row = list_val.at(0);
        OZ (part_vals.push_back(row));
        OZ (part_ids.push_back(partition->get_part_id()));
      } else {
        ret = OB_ERR_UNEXPECTED;
        LOG_WARN("external table partition list value num should be one", K(ret));
      }
    }
  }
  return ret;
}

int ObExternalTableFileManager::calculate_odps_part_val_by_part_spec(const ObTableSchema *table_schema,
                                                                     const ObIArray<ObExternalFileInfoTmp> &file_infos,
                                                                     ObIArray<ObNewRow> &part_vals,
                                                                     ObIAllocator &allocator)
{
  int ret = OB_SUCCESS;
  bool is_odps_external_table = false;
  if (OB_FAIL(ObSQLUtils::is_odps_external_table(table_schema, is_odps_external_table))) {
    LOG_WARN("failed to check is odps external table or not", K(ret));
  } else if (is_odps_external_table) {
    const common::ObPartitionKeyInfo &part_key_info = table_schema->get_partition_key_info();
    const int part_key_size = part_key_info.get_size();
    for (int64_t i = 0; OB_SUCC(ret) && i < file_infos.count(); i++) {
      const ObString &all_part_spec = file_infos.at(i).file_url_;
      ObSEArray<ObString, 6> part_spec_list;
      if (OB_FAIL(ObSQLUtils::extract_odps_part_spec(all_part_spec, part_spec_list))) {
        LOG_WARN("failed to extract odps part spec", K(ret), K(all_part_spec));
      } else if (part_spec_list.count() != part_key_size) {
        LOG_WARN("unexpected count find part spec of odps", K(ret), K(file_infos), K(file_infos.count()), K(i), K(all_part_spec), K(part_spec_list.count()), K(part_key_size));
        ret = OB_EXTERNAL_ODPS_UNEXPECTED_ERROR;
        LOG_WARN("unexpected count find part spec of odps", K(ret), K(file_infos), K(file_infos.count()), K(i), K(all_part_spec), K(part_spec_list.count()), K(part_key_size));
        LOG_USER_ERROR(OB_EXTERNAL_ODPS_UNEXPECTED_ERROR, "unexpected count of partition key between odps table and external table");
      } else {
        ObNewRow odps_part_row;
        ObObj *obj_array = nullptr;
        if (OB_ISNULL(obj_array = static_cast<ObObj*>(allocator.alloc(sizeof(ObObj) * part_key_size)))) {
          ret = OB_ALLOCATE_MEMORY_FAILED;
          LOG_WARN("fail to allocate memory", K(ret));
        } else {
          for (ObObj *ptr = obj_array; OB_SUCC(ret) && ptr < obj_array + part_key_size; ++ptr) {
            new(ptr)ObObj();
          }
          odps_part_row.assign(obj_array, part_key_size);
        }
        for (int64_t j = 0; OB_SUCC(ret) && j < part_spec_list.count(); ++j) {
          const ObRowkeyColumn *part_col = part_key_info.get_column(j);
          ObObjType part_key_type = ObUnknownType;
          ObString& part_spec = part_spec_list.at(j);
          if (OB_ISNULL(part_col)) {
            ret = OB_ERR_UNEXPECTED;
            LOG_WARN("unexpected null ptr", K(ret), K(j));
          } else if (FALSE_IT(part_key_type = part_col->get_meta_type().get_type())) {
          } else if (part_key_type == ObVarcharType ||
                     part_key_type == ObCharType ||
                     part_key_type == ObMediumTextType) {
            oceanbase::common::ObObjMeta meta_type = part_col->get_meta_type();
            ObCollationType coll_dst = static_cast<ObCollationType>(meta_type.get_cs_type());
            ObCollationType coll_src = CS_TYPE_UTF8MB4_BIN;
            int64_t dst_maxblen = 0;
            int64_t src_minblen = 0;
            if (OB_FAIL(ObCharset::get_mbmaxlen_by_coll(coll_dst, dst_maxblen))) {
              LOG_WARN("failed to get dst mb max len", K(ret), K(coll_dst));
            } else if (OB_FAIL(ObCharset::get_mbminlen_by_coll(coll_src, src_minblen))) {
              LOG_WARN("failed to get src mb min len", K(ret), K(coll_src));
            } else {
              void *dst_buf = NULL;
              uint64_t dst_buf_size = (part_spec.length() / src_minblen) * dst_maxblen;
              uint32_t dst_len = 0;
              if (OB_ISNULL(dst_buf = allocator.alloc(dst_buf_size))) {
                ret = OB_ALLOCATE_MEMORY_FAILED;
                LOG_WARN("failed to alloc buf", K(ret));
              } else if (OB_FAIL(ObCharset::charset_convert(coll_src, part_spec.ptr(), part_spec.length(), coll_dst, static_cast<char*>(dst_buf), dst_buf_size, dst_len))) {
                LOG_WARN("failed to convert charset", K(ret));
              } else if (part_key_type == ObMediumTextType) { // string type
                ObString lob_data(static_cast<int64_t>(dst_len), static_cast<char*>(dst_buf));
                ObString lob_with_header;
                if (OB_FAIL(ObLobManager::fill_lob_header(allocator, lob_data, lob_with_header))) {
                  LOG_WARN("failed to fill lob header");
                } else {
                  odps_part_row.get_cell(j).set_meta_type(part_col->get_meta_type());
                  odps_part_row.get_cell(j).set_string(ObMediumTextType, lob_with_header.ptr(), lob_with_header.length());
                  odps_part_row.get_cell(j).set_has_lob_header();
                }
              } else {
                odps_part_row.get_cell(j).set_meta_type(part_col->get_meta_type());
                odps_part_row.get_cell(j).set_varchar_value(static_cast<char*>(dst_buf),
                                                            static_cast<int64_t>(dst_len));
              }
            }
          } else if (part_key_type == ObTinyIntType ||
                     part_key_type == ObSmallIntType ||
                     part_key_type == ObMediumIntType ||
                     part_key_type == ObInt32Type ||
                     part_key_type == ObIntType) {
            int64_t val = 0;
            for (int64_t k = 0; OB_SUCC(ret) && k < part_spec.length(); ++k) {
              if (part_spec.ptr()[k] >= '0' &&  part_spec.ptr()[k] <= '9') {
                val = val * 10 + part_spec.ptr()[k] - '0';
              } else {
                ret = OB_ERR_UNEXPECTED;
                LOG_WARN("unexpected part_spec", K(ret), K(part_spec));
              }
            }
            if (OB_SUCC(ret)) {
              odps_part_row.get_cell(j).set_meta_type(part_col->get_meta_type());
              odps_part_row.get_cell(j).set_int(val);
            }
          } else if (is_oracle_mode() && ObNumberType == part_key_type) {
            number::ObNumber num;
            if (OB_FAIL(num.from(part_spec.ptr(), part_spec.length(), allocator))) {
              LOG_WARN("cast string to number failed", K(ret), K(part_spec),
                       K(part_key_type));
            } else {
              LOG_INFO("cast string to number success", K(ret));
              odps_part_row.get_cell(j).set_meta_type(part_col->get_meta_type());
              odps_part_row.get_cell(j).set_number(num);
            }
          } else {
            // TODO(bitao): fix support to run in oracle mode and more types
            ret = OB_ERR_UNEXPECTED;
            LOG_WARN("unexpected part_key_type", K(part_spec), K(part_key_type),
                     K(j), K(ret));
          }
        }
        if (OB_SUCC(ret) && OB_FAIL(part_vals.push_back(odps_part_row))) {
          LOG_WARN("failed push back odps_part_row", K(ret));
        }
      }
    }
  }
  return ret;
}

int ObExternalTableFileManager::calculate_file_part_val_by_file_name(const ObTableSchema *table_schema,
                                                                const ObIArray<ObExternalFileInfoTmp> &file_infos,
                                                                ObIArray<ObNewRow> &part_vals,
                                                                share::schema::ObSchemaGetterGuard &schema_guard,
                                                                ObExecContext &exec_ctx)
{
  int ret = OB_SUCCESS;
  ObIAllocator &allocator = exec_ctx.get_allocator();
  ObArray<sql::ObTempExpr *> temp_exprs;
  ObNewRow file_name_row; //only store file name, auto-part list val will only be calc by file name.
  CK (OB_NOT_NULL(table_schema) && OB_LIKELY(table_schema->is_external_table()));
  ObString file_location;
  OZ (ObExternalTableUtils::get_external_file_location(*table_schema, schema_guard, exec_ctx.get_allocator(), file_location));
  const bool is_local_storage = ObSQLUtils::is_external_files_on_local_disk(file_location);
  OZ (cg_partition_expr_rt_expr(table_schema, exec_ctx.get_expr_factory(), exec_ctx.get_my_session(),
                                schema_guard, exec_ctx.get_allocator(), temp_exprs));
  OZ (build_row_for_file_name(file_name_row, exec_ctx.get_allocator()));
  bool is_odps_external_table = false;
  if (OB_FAIL(ret)) {
  } else if (OB_FAIL(ObSQLUtils::is_odps_external_table(table_schema, is_odps_external_table))) {
    LOG_WARN("failed to check is odps external table or not", K(ret));
  }
  for (int64_t i = 0; OB_SUCC(ret) && i < file_infos.count(); i++) {
    ObNewRow list_val;
    ObObj *obj_array = nullptr;
    if (file_name_row.get_count() > 0) {
      file_name_row.get_cell(0).set_string(ObVarcharType, is_odps_external_table ?
                                                            file_infos.at(i).file_url_.after(equals_delimiter) :
                                                            (is_local_storage ?
                                                              file_infos.at(i).file_url_.after(ip_delimiter) :
                                                              file_infos.at(i).file_url_));
    } else {
      ret = OB_ERR_UNEXPECTED;
      LOG_WARN("row cell count not expected", K(ret));
    }
    if (OB_FAIL(ret)) {
    } else if (OB_ISNULL(obj_array = static_cast<ObObj*>(
                allocator.alloc(sizeof(ObObj) * temp_exprs.count())))) {
      ret = OB_ALLOCATE_MEMORY_FAILED;
      LOG_WARN("fail to allocate memory", K(ret));
    } else {
      for (ObObj *ptr = obj_array; ptr < obj_array + temp_exprs.count(); ++ptr) {
        new(ptr)ObObj();
        ptr->set_collation_type(ObCharset::get_system_collation());
      }
      if (OB_SUCC(ret)) {
        list_val.assign(obj_array, temp_exprs.count());
      }
    }
    for (int64_t j = 0; OB_SUCC(ret) && j < temp_exprs.count(); j++) {
      OZ (temp_exprs.at(j)->eval(exec_ctx, file_name_row, list_val.get_cell(j)));
    }
    OZ (part_vals.push_back(list_val));
  }
  return ret;
}

int ObExternalTableFileManager::add_item_to_map(ObIAllocator &allocator,
                                                common::hash::ObHashMap<int64_t, ObArray<ObExternalFileInfoTmp> *> &hash_map,
                                                int64_t part_id, const ObExternalFileInfoTmp &file_info)
{
  int ret = OB_SUCCESS;
  ObArray<ObExternalFileInfoTmp> *part_file_urls = NULL;
  ret = hash_map.get_refactored(part_id, part_file_urls);
  if (ret == OB_HASH_NOT_EXIST) {
    ret = OB_SUCCESS;
    if (OB_ISNULL(part_file_urls = OB_NEWx(ObArray<ObExternalFileInfoTmp>, &allocator))) {
      ret = OB_ALLOCATE_MEMORY_FAILED;
      LOG_WARN("fail to new array", KR(ret));
    } else if (OB_FAIL(hash_map.set_refactored(part_id, part_file_urls))) {
      LOG_WARN("fail to set refactored", KR(ret), K(part_id));
    }
    if (OB_FAIL(ret)) {
      if (nullptr != part_file_urls) {
        part_file_urls->~ObArray<ObExternalFileInfoTmp>();
        allocator.free(part_file_urls);
        part_file_urls = NULL;
      }
    }
  } else {
    LOG_WARN("hash map get key value failed", K(ret));
  }
  if (OB_SUCC(ret)) {
    OZ (part_file_urls->push_back(file_info));
  }
  return ret;
}

int ObExternalTableFileManager::calculate_all_files_partitions(share::schema::ObSchemaGetterGuard &schema_guard,
                                                             ObExecContext &exec_ctx,
                                                             const ObTableSchema *table_schema,
                                                             const ObIArray<ObExternalFileInfoTmp> &file_infos,
                                                             ObIArray<int64_t> &file_part_ids,
                                                             ObIArray<ObNewRow> &partitions_to_add,
                                                             ObIArray<ObPartition *> &partitions_to_del)
{
  int ret = OB_SUCCESS;
  int64_t mock_part_id = 0;
  ObArray<ObNewRow> existed_part_vals;
  ObArray<int64_t> existed_part_ids;
  ObArray<ObNewRow> file_part_vals;
  CK (OB_NOT_NULL(table_schema) && OB_LIKELY(table_schema->is_external_table()));
  OZ (get_all_partition_list_val(table_schema, existed_part_vals, existed_part_ids));
  bool is_odps_external_table = false;
  if (OB_FAIL(ret)) {
  } else if (OB_FAIL(ObSQLUtils::is_odps_external_table(table_schema, is_odps_external_table))) {
    LOG_WARN("failed to check is odps external table or not", K(ret));
  } else if (is_odps_external_table) {
    OZ(calculate_odps_part_val_by_part_spec(table_schema, file_infos, file_part_vals, exec_ctx.get_allocator()));
  } else {
    OZ (calculate_file_part_val_by_file_name(table_schema, file_infos, file_part_vals, schema_guard, exec_ctx));
  }
  for (int64_t i = 0; OB_SUCC(ret) && i < file_part_vals.count(); i++) {
    int64_t idx = -1;
    OZ (find_partition_existed(existed_part_vals, file_part_vals.at(i), idx));
    if (OB_FAIL(ret)) {
    } else if (idx == -1) {
      OZ (existed_part_vals.push_back(file_part_vals.at(i)));
      OZ (existed_part_ids.push_back(mock_part_id));
      OZ (file_part_ids.push_back(mock_part_id));
      OZ (partitions_to_add.push_back(file_part_vals.at(i)));
      mock_part_id--;
    } else if (idx >= 0 && idx < existed_part_ids.count()) {
      OZ (file_part_ids.push_back(existed_part_ids.at(idx)));
    } else {
      ret = OB_ERR_UNEXPECTED;
      LOG_WARN("find item failed", K(ret), K(idx));
    }
  }
  for (int64_t i = 0; OB_SUCC(ret) && i < table_schema->get_partition_num(); i++) {
    bool found = false;
    ObPartition *partition = table_schema->get_part_array()[i];
    if (OB_ISNULL(partition)) {
      ret = OB_ERR_UNEXPECTED;
      LOG_WARN("partition is null", K(ret));
    } else if (partition->get_part_name().compare("P_DEFAULT") == 0) {
      found = true;
    }
    for (int64_t j = 0; OB_SUCC(ret) && !found && j < file_part_ids.count(); j++) {
      if (partition->get_part_id() == file_part_ids.at(j)) {
        found = true;
      }
    }
    if (OB_SUCC(ret) && !found) {
      OZ (partitions_to_del.push_back(partition));
    }
  }
  return ret;
}

int ObExternalTableFileManager::update_inner_table_file_list(
    sql::ObExecContext &exec_ctx,
    const uint64_t tenant_id,
    const uint64_t table_id,
    ObArray<share::ObExternalTableBasicFileInfo> &basic_file_infos,
    ObIArray<uint64_t> &updated_part_ids,
    bool &has_partition_changed,
    const uint64_t part_id,
    bool collect_statistic)
{
  int ret = OB_SUCCESS;
  ObMySQLTransaction trans;
  ObArenaAllocator allocator;
  share::schema::ObSchemaGetterGuard schema_guard;
  const ObTableSchema *table_schema = NULL;
  OZ (GCTX.schema_service_->get_tenant_schema_guard(tenant_id, schema_guard));
  OZ (schema_guard.get_table_schema(tenant_id, table_id, table_schema));
  CK (OB_NOT_NULL(table_schema));
  CK (OB_NOT_NULL(GCTX.sql_proxy_),
      OB_NOT_NULL(GCTX.schema_service_));
  OZ (trans.start(GCTX.sql_proxy_, tenant_id));
  OZ (lock_for_refresh(trans, tenant_id, table_id));
  ObArray<ObExternalFileInfoTmp> file_infos;
  for (int64_t i = 0; OB_SUCC(ret) && i < basic_file_infos.count(); i++) {
    OZ(file_infos.push_back(ObExternalFileInfoTmp(basic_file_infos.at(i).url_,
                                                  basic_file_infos.at(i).content_digest_,
                                                  basic_file_infos.at(i).size_,
                                                  basic_file_infos.at(i).last_modify_time_,
                                                  part_id)));
  }
  if (OB_FAIL(ret)) {
  } else if (part_id != -1) {
    OZ (update_inner_table_files_list_by_part(trans, tenant_id, table_id, part_id, file_infos, updated_part_ids));
  } else {
    OZ (update_inner_table_files_list_by_table(exec_ctx, trans, tenant_id, table_id, file_infos, updated_part_ids, has_partition_changed));
  }
  if (OB_FAIL(ret)) {
  } else if (OB_FAIL(collect_odps_table_statistics(collect_statistic, tenant_id, table_id, updated_part_ids, trans))) {
    LOG_WARN("failed to collect odps table statistics", K(collect_statistic), K(tenant_id), K(table_id));
  }
  OZ (trans.end(true));
  if (trans.is_started()) {
    trans.end(false);
  }
  return ret;
}

int ObExternalTableFileManager::collect_odps_table_statistics(const bool collect_statistic,
                                                              const uint64_t tenant_id,
                                                              const uint64_t table_id,
                                                              ObIArray<uint64_t> &updated_part_ids,
                                                              ObMySQLTransaction &trans)
{
  int ret = OB_SUCCESS;
  bool is_odps_external_table = false;
  if (updated_part_ids.empty()) {
    // do nothing
  } else if (OB_FAIL(ObSQLUtils::is_odps_external_table(tenant_id, table_id, is_odps_external_table))) {
    LOG_WARN("failed to check is odps table or not", K(ret), K(tenant_id), K(table_id));
  } else if (is_odps_external_table && collect_statistic) {
    int64_t update_rows = 0;
    ObSqlString update_sql;
    int64_t dop_of_collect_external_table_statistics = 1;
    omt::ObTenantConfigGuard tenant_config(TENANT_CONF(tenant_id));
    if (OB_LIKELY(tenant_config.is_valid()) && static_cast<int64_t>(tenant_config->_dop_of_collect_external_table_statistics) > 0) {
      dop_of_collect_external_table_statistics = tenant_config->_dop_of_collect_external_table_statistics;
    } else {
      int64_t default_dop = (updated_part_ids.count() / 4 > 0) ? (updated_part_ids.count() / 4) : 1;
      double min_cpu;
      double max_cpu;
      if (OB_ISNULL(GCTX.omt_)) {
        ret = OB_ERR_UNEXPECTED;
      } else if (OB_FAIL(GCTX.omt_->get_tenant_cpu(tenant_id, min_cpu, max_cpu))) {
        LOG_WARN("fail to get tenant cpu", K(ret));
      } else {
        dop_of_collect_external_table_statistics = std::min(updated_part_ids.count() / 4, static_cast<int64_t>(max_cpu));
      }
    }
    OZ(update_sql.assign_fmt("UPDATE /*+ enable_parallel_dml parallel(%ld) */ %s SET FILE_SIZE = CALC_ODPS_SIZE(FILE_URL, TABLE_ID) WHERE TABLE_ID = %ld and PART_ID IN (",
                            dop_of_collect_external_table_statistics,
                            OB_ALL_EXTERNAL_TABLE_FILE_TNAME,
                            table_id));
    for (int64_t i = 0; OB_SUCC(ret) && i < updated_part_ids.count(); i++) {
      OZ(update_sql.append_fmt("%ld%c", updated_part_ids.at(i), ((updated_part_ids.count() - 1) == i) ? ')' : ','));
    }
    OZ(trans.write(tenant_id, update_sql.ptr(), update_rows));
  }
  return ret;
}

int ObExternalTableFileManager::get_file_sizes_by_map(ObIArray<ObString> &file_urls,
                                                      ObIArray<int64_t> &file_sizes,
                                                      common::hash::ObHashMap<ObString, int64_t> &map)
{
  int ret = OB_SUCCESS;
  file_sizes.reuse();
  for (int64_t i = 0; OB_SUCC(ret) && i < file_urls.count(); i++) {
    int64_t size = 0;
    if (OB_FAIL(map.get_refactored(file_urls.at(i), size))) {
      LOG_WARN("get file url to size failed", K(ret));
    } else if (OB_FAIL(file_sizes.push_back(size))) {
      LOG_WARN("push back failed", K(ret));
    }
  }
  return ret;
}

int ObExternalTableFileManager::update_inner_table_files_list_by_table(
    sql::ObExecContext &exec_ctx,
    ObMySQLTransaction &trans,
    const uint64_t tenant_id,
    const uint64_t table_id,
    const ObIArray<ObExternalFileInfoTmp> &file_infos,
    ObIArray<uint64_t> &updated_part_ids,
    bool &has_partition_changed)
{
  int ret = OB_SUCCESS;
  ObMySQLProxy *sql_proxy = GCTX.sql_proxy_;
  ObArenaAllocator allocator;
  CK (OB_NOT_NULL(sql_proxy),
      OB_NOT_NULL(GCTX.schema_service_));
  share::schema::ObSchemaGetterGuard schema_guard;
  const ObTableSchema *table_schema = NULL;
  const ObDatabaseSchema *database_schema = NULL;
  OZ (GCTX.schema_service_->get_tenant_schema_guard(tenant_id, schema_guard));
  OZ (schema_guard.get_table_schema(tenant_id, table_id, table_schema));
  CK (OB_NOT_NULL(table_schema));
  OZ (schema_guard.get_database_schema(tenant_id, table_schema->get_database_id(), database_schema));
  CK (OB_NOT_NULL(database_schema));
  if (OB_FAIL(ret)) {
  } else if (!table_schema->is_partitioned_table()) {
    OZ (update_inner_table_files_list_by_part(trans, tenant_id, table_id, table_id, file_infos, updated_part_ids));
  } else {
    int64_t max_part_id = 0;
    common::hash::ObHashMap<int64_t, ObArray<ObExternalFileInfoTmp> *> part_id_to_file_urls; //part id to file urls.
    if (OB_SUCC(ret)) {
      OZ (part_id_to_file_urls.create(file_infos.count() + 1, "ExtFileUrl")); //todo: too large

      int64_t partition_to_add_num = 0;
      ObArray<int64_t> part_ids;
      ObArray<ObNewRow> partitions_to_add;
      ObArray<ObPartition *> partitions_to_del;
      OZ (calculate_all_files_partitions(schema_guard, exec_ctx, table_schema, file_infos, part_ids, partitions_to_add, partitions_to_del));
      if (OB_SUCC(ret)) {
        int64_t max_part_idx = 0;
        OZ (table_schema->get_max_part_idx(max_part_idx, true/*without default*/));

        if (OB_SUCC(ret) && partitions_to_del.count() > 0) {
          ObAlterTableStmt *alter_table_stmt = NULL;
          OZ (create_alter_table_stmt(exec_ctx, table_schema, database_schema, partitions_to_del.count(), ObAlterTableArg::DROP_PARTITION, alter_table_stmt));
          if (OB_SUCC(ret)) {
            for (int64_t i = 0; OB_SUCC(ret) && i < partitions_to_del.count(); i++) {
              if (OB_ISNULL(partitions_to_del.at(i))) {
                ret = OB_ERR_UNEXPECTED;
                LOG_WARN("partitions to del is null", K(ret));
              } else if (OB_FAIL(alter_table_stmt->get_alter_table_arg().alter_table_schema_.add_partition(*partitions_to_del.at(i)))) {
                LOG_WARN("failed to add partition", K(ret));
              }
            }
          }
          ObArray<ObExternalFileInfoTmp> empty;
          // Will not delete the partition now.
          // ObArray<int64_t> file_part_ids_del;
          // ObSEArray<ObAddr, 8> all_servers;
          // OZ (GCTX.location_service_->external_table_get(tenant_id, table_id, all_servers));
          // OZ (alter_partition_for_ext_table(trans, exec_ctx, alter_table_stmt, file_part_ids_del));
          for (int i = 0; OB_SUCC(ret) && i < partitions_to_del.count(); i++) {
            if (OB_ISNULL(partitions_to_del.at(i))) {
              ret = OB_ERR_UNEXPECTED;
              LOG_WARN("partitions to del is null", K(ret));
            } else {
              OZ (update_inner_table_files_list_by_part(trans, tenant_id, table_id, partitions_to_del.at(i)->get_part_id(), empty, updated_part_ids));
            }
          }
        }
        if (OB_SUCC(ret) && partitions_to_add.count() > 0) {
          has_partition_changed = true;
          ObAlterTableStmt *alter_table_stmt = NULL;
          OZ (create_alter_table_stmt(exec_ctx, table_schema, database_schema, partitions_to_add.count(), ObAlterTableArg::ADD_PARTITION, alter_table_stmt));
          for (int64_t i = 0; OB_SUCC(ret) && i < partitions_to_add.count(); i++) {
            ObSqlString tmp_part_name;
            ObString part_name;
            OZ (tmp_part_name.append_fmt("P%ld", ++max_part_idx));
            OZ (ob_write_string(exec_ctx.get_allocator(), tmp_part_name.string(), part_name));
            OZ (add_partition_for_alter_stmt(alter_table_stmt, part_name, partitions_to_add.at(i)));
          }
          //ObFixedArray<int64_t, ObIAllocator> file_part_ids_added(&exec_ctx.get_allocator(), partitions_to_add.count());
          ObArray<int64_t> file_part_ids_added;
          OZ (alter_partition_for_ext_table(trans, exec_ctx, alter_table_stmt, file_part_ids_added));
          #define IS_EXT_MOCK_PART_ID(id) id <= 0 ? true : false
          #define GET_EXT_MOCK_PART_ID(id) -id
          #define CHECK_EXT_MOCK_PART_ID_VALID(real_part_ids, id) if (OB_UNLIKELY(id < 0 \
                                                                            && id >= real_part_ids.count())) {\
                                                                    ret = OB_ERR_UNEXPECTED; \
                                                                  }
          #define MAP_EXT_MOCK_PART_ID_TO_REAL_PART_ID(real_part_ids, id) real_part_ids.at(id)
          //make mock_part_id to real part id;
          for (int64_t i = 0; OB_SUCC(ret) && i < part_ids.count(); i++) {
            int64_t part_id = 0;
            if (IS_EXT_MOCK_PART_ID(part_ids.at(i))) {
              int64_t mock_part_id = GET_EXT_MOCK_PART_ID(part_ids.at(i));
              CHECK_EXT_MOCK_PART_ID_VALID(file_part_ids_added, mock_part_id);
              part_id = MAP_EXT_MOCK_PART_ID_TO_REAL_PART_ID(file_part_ids_added, mock_part_id);
              part_ids.at(i) = part_id;
            }
          }
          #undef GET_EXT_MOCK_PART_ID
          #undef IS_EXT_MOCK_PART_ID
        }
      }

      for (int64_t i = 0; OB_SUCC(ret) && i < part_ids.count(); i++) {
        OZ (add_item_to_map(exec_ctx.get_allocator(), part_id_to_file_urls, part_ids.at(i), file_infos.at(i)));
      }
      //OZ (get_part_id_to_file_urls_map(table_schema, database_schema, is_local_storage, file_urls, file_sizes, schema_guard, exec_ctx, trans, part_id_to_file_urls));
      for (common::hash::ObHashMap<int64_t, ObArray<ObExternalFileInfoTmp> *>::iterator it = part_id_to_file_urls.begin();
          OB_SUCC(ret) && it != part_id_to_file_urls.end(); it++) {
        CK (OB_NOT_NULL(it->second));
        //OZ (get_file_sizes_by_map(*it->second, mapped_sizes, file_urls_to_sizes));
        OZ (update_inner_table_files_list_by_part(trans, tenant_id, table_id, it->first, *it->second, updated_part_ids));
      }
      LOG_TRACE("external table refresh report", "total parts", part_ids.count(), "total files", file_infos.count(),
                "partitions to add", partitions_to_add.count(), "partitions to del", partitions_to_del.count(),
                K(has_partition_changed));
    }

    for (common::hash::ObHashMap<int64_t, ObArray<ObExternalFileInfoTmp> *>::iterator it = part_id_to_file_urls.begin(); it != part_id_to_file_urls.end(); it++) {
      if (nullptr != it->second) {
        it->second->~ObArray<ObExternalFileInfoTmp>();
        exec_ctx.get_allocator().free(it->second);
        it->second = nullptr;
      }
    }
  }
  return ret;
}

int ObExternalTableFileManager::get_external_file_list_on_device(
    const ObString &location,
    const ObString &pattern,
    const ObExprRegexpSessionVariables &regexp_vars,
    ObIArray<ObString> &file_urls,
    ObIArray<int64_t> &file_sizes,
    ObIArray<int64_t> &last_modify_times,
    ObIArray<ObString> &content_digests,
    const ObString &access_info,
    ObIAllocator &allocator)
{
  int ret = OB_SUCCESS;
  ObExternalFileInfoCollector collector(allocator);
  if (OB_FAIL(collector.init(location, access_info))) {
    LOG_WARN("failed to init", K(ret));
  } else if (OB_FAIL(collector.get_file_list(location, pattern, regexp_vars, file_urls, file_sizes,
                                             last_modify_times, content_digests))) {
    LOG_WARN("failed to get file list", K(ret));
  }
  return ret;
}

int ObExternalTableFileManager::collect_files_content_digest(
  const common::ObString &location, const common::ObString &access_info,
  const common::ObIArray<common::ObString> &file_urls, common::ObIAllocator &allocator,
  common::ObIArray<ObString> &content_digests)
{
  int ret = OB_SUCCESS;
  ObExternalFileInfoCollector collector(allocator);
  if (OB_FAIL(collector.init(location, access_info))) {
    LOG_WARN("failed to init", K(ret));
  } else if (OB_FAIL(collector.collect_files_content_digest(file_urls, content_digests))) {
    LOG_WARN("failed to get file list", K(ret));
  }
  return ret;
}

int ObExternalTableFileManager::get_external_file_list_on_device_with_cache(
    const ObString &location,
    const uint64_t tenant_id,
    const uint64_t modify_ts,
    const ObString &pattern,
    const ObExprRegexpSessionVariables &regexp_vars,
    ObExternalTableFiles &external_table_files,
    const ObString &access_info,
    ObIAllocator &allocator,
    int64_t refresh_interval_ms)
{
  int ret = OB_SUCCESS;
  const ObExternalTableFiles *table_files_from_cache = NULL;

  if (refresh_interval_ms > 0) {
    ObExternalTableFiles tmp_file_list;
    ObKVCacheHandle handle;
    ObExternalTableFileListKey key;
    key.tenant_id_ = tenant_id;
    key.path_ = location;
    if (OB_FAIL(external_file_list_cache_.get(key, table_files_from_cache, handle))) {
      if (OB_ENTRY_NOT_EXIST != ret) {
        LOG_WARN("fail to get from external_file_list_cache", K(ret), K(key));
      }
    }

    if ((OB_SUCC(ret)
         && (modify_ts != table_files_from_cache->modify_ts_
             || is_cache_value_timeout(table_files_from_cache->create_ts_, refresh_interval_ms)))
        || OB_ENTRY_NOT_EXIST == ret) {
      // only one worker need do the job
      int64_t bucket_id = key.hash() % LOAD_CACHE_LOCK_CNT;
      int64_t total_wait_secs = 0;

      // lock
      while (OB_FAIL(fill_cache_locks_[bucket_id].lock(LOCK_TIMEOUT)) && OB_TIMEOUT == ret
             && OB_SUCC(THIS_WORKER.check_status())) {
        total_wait_secs += (LOCK_TIMEOUT / 1000000);
        LOG_WARN("fill external table cache wait", K(ret), K(total_wait_secs));
      }

      // fetch again
      if (OB_FAIL(external_file_list_cache_.get(key, table_files_from_cache, handle))) {
        if (OB_ENTRY_NOT_EXIST != ret) {
          LOG_WARN("fail to get from external_file_list_cache", K(ret), K(key));
        }
      }

      if ((OB_SUCC(ret)
           && (modify_ts != table_files_from_cache->modify_ts_
               || is_cache_value_timeout(table_files_from_cache->create_ts_, refresh_interval_ms)))
          || OB_ENTRY_NOT_EXIST == ret) {
        if (OB_FAIL(get_external_file_list_on_device(location,
                                                     modify_ts,
                                                     pattern,
                                                     regexp_vars,
                                                     access_info,
                                                     allocator,
                                                     tmp_file_list))) {
          LOG_WARN("fail to get from external file list",
                   K(ret),
                   K(key),
                   K(location),
                   K(modify_ts),
                   K(pattern),
                   K(access_info));
        } else if (OB_FAIL(external_file_list_cache_.put_and_fetch(key,
                                                                   tmp_file_list,
                                                                   table_files_from_cache,
                                                                   handle,
                                                                   true))) {
          LOG_WARN("fail to put and get external file list to cache", K(ret), K(key));
        }
      }

      // unlock
      if (fill_cache_locks_[bucket_id].self_locked()) {
        fill_cache_locks_[bucket_id].unlock();
      }
    }

    if (OB_SUCC(ret)) {
      if (OB_NOT_NULL(table_files_from_cache)) {
        external_table_files.file_urls_.assign(table_files_from_cache->file_urls_);
        external_table_files.file_sizes_.assign(table_files_from_cache->file_sizes_);
        external_table_files.modify_times_.assign(table_files_from_cache->modify_times_);
      } else {
        LOG_WARN("fail to get external files from cache", K(ret), K(key));
      }
    }
  } else {
    table_files_from_cache = new ObExternalTableFiles;
    if (OB_FAIL(get_external_file_list_on_device(location,
                                                 modify_ts,
                                                 pattern,
                                                 regexp_vars,
                                                 access_info,
                                                 allocator,
                                                 external_table_files))) {
      LOG_WARN("fail to get from external file list",
               K(ret),
               K(location),
               K(modify_ts),
               K(pattern),
               K(access_info));
    }
  }

  LOG_TRACE("get files from cache", K(external_table_files));
  return ret;
}

int ObExternalTableFileManager::get_external_file_list_on_device(
    const ObString &location,
    const uint64_t modify_ts,
    const ObString &pattern,
    const ObExprRegexpSessionVariables &regexp_vars,
    const ObString &access_info,
    ObIAllocator &allocator,
    ObExternalTableFiles &file_list)
{
  int ret = OB_SUCCESS;
  ObArray<int64_t> tmp_file_sizes;
  ObArray<int64_t> tmp_last_modify_times;
  ObArray<ObString> tmp_content_digests;
  ObArray<ObString> tmp_file_urls;

  if (OB_FAIL(get_external_file_list_on_device(location,
                                               pattern,
                                               regexp_vars,
                                               tmp_file_urls,
                                               tmp_file_sizes,
                                               tmp_last_modify_times,
                                               tmp_content_digests,
                                               access_info,
                                               allocator))) {
    LOG_WARN("fail to get from external file list", K(ret), K(location));
  } else {
    file_list.modify_ts_ = modify_ts;
    file_list.create_ts_ = ObTimeUtil::current_time_ms();
    if (tmp_file_urls.count() != tmp_file_sizes.count()
        || tmp_file_sizes.count() != tmp_last_modify_times.count()) {
      ret = OB_ERR_UNEXPECTED;
      LOG_WARN("fail to get from external file list", K(ret), K(location));
    } else if (tmp_file_urls.count() > 0) {
      OZ(file_list.file_urls_.allocate_array(allocator, tmp_file_urls.count()));
      OZ(file_list.file_sizes_.allocate_array(allocator, tmp_file_sizes.count()));
      OZ(file_list.modify_times_.allocate_array(allocator, tmp_last_modify_times.count()));
    }

    for (int64_t i = 0; OB_SUCC(ret) && i < tmp_file_urls.count(); ++i) {
      ObSqlString t;
      OZ(t.append_fmt("%.*s/%.*s",
                      location.length(),
                      location.ptr(),
                      tmp_file_urls.at(i).length(),
                      tmp_file_urls.at(i).ptr()));
      OZ(ob_write_string(allocator, t.string(), file_list.file_urls_.at(i)));
      file_list.file_sizes_.at(i) = tmp_file_sizes.at(i);
      file_list.modify_times_.at(i) = tmp_last_modify_times.at(i);
    }
  }

  return ret;
}

int ObExternalTableFileManager::get_partitions_info_with_cache(const ObTableSchema &table_schema,
                                                               ObSqlSchemaGuard &sql_schema_guard,
                                                               common::ObIAllocator& allocator,
                                                               int64_t refresh_interval_ms,
                                                               ObArray<PartitionInfo*> &partition_infos)
{
  int ret = OB_SUCCESS;
  const share::ObILakeTableMetadata *metadata
      = sql_schema_guard.get_table_metadata(table_schema.get_table_id());
  ObArenaAllocator tmp_allocator;
  ObCachedCatalogMetaGetter catalog_meta_getter{*sql_schema_guard.get_schema_guard(),
                                                tmp_allocator};
  Partitions partitions;
  const ObExternalTablePartitions *table_partitions = NULL;
  ObKVCacheHandle handle;
  if (OB_ISNULL(metadata)) {
    ret = OB_ERR_UNEXPECTED;
    LOG_WARN("get unexpected null metadata", K(ret), K(table_schema));
  } else if (refresh_interval_ms > 0) {
    // get partition info from cache
    ObExternalTablePartitionsKey key;
    key.tenant_id_ = table_schema.get_tenant_id();
    key.catalog_id_ = table_schema.get_catalog_id();
    key.db_name_ = metadata->namespace_name_;
    key.tb_name_ = metadata->table_name_;
    if (OB_FAIL(external_partitions_cache_.get(key, table_partitions, handle))) {
      if (OB_ENTRY_NOT_EXIST != ret) {
        LOG_WARN("fail to get from external_partitions_cache", K(ret), K(key));
      }
    }

    if ((OB_SUCC(ret) && is_cache_value_timeout(table_partitions->create_ts_, refresh_interval_ms))
        || OB_ENTRY_NOT_EXIST == ret) {
      // only one worker need do the job
      int64_t bucket_id = key.hash() % LOAD_CACHE_LOCK_CNT;
      int64_t total_wait_secs = 0;

      // lock
      while (OB_FAIL(fill_cache_locks_[bucket_id].lock(LOCK_TIMEOUT)) && OB_TIMEOUT == ret
             && OB_SUCC(THIS_WORKER.check_status())) {
        total_wait_secs += (LOCK_TIMEOUT / 1000000);
        LOG_WARN("fill external table cache wait", K(ret), K(total_wait_secs));
      }

      // fetch again
      if (OB_FAIL(external_partitions_cache_.get(key, table_partitions, handle))) {
        if (OB_ENTRY_NOT_EXIST != ret) {
          LOG_WARN("fail to get from external_file_list_cache", K(ret), K(key));
        }
      }

      if ((OB_SUCC(ret)
           && is_cache_value_timeout(table_partitions->create_ts_, refresh_interval_ms))
          || OB_ENTRY_NOT_EXIST == ret) {
        ret = OB_SUCCESS;
        ObExternalTablePartitions pts;
        if (OB_FAIL(get_partitions_info(table_schema, sql_schema_guard, allocator, pts))) {
          LOG_WARN("fail to get partitions info", K(ret), K(table_schema));
        } else if (OB_FAIL(external_partitions_cache_
                               .put_and_fetch(key, pts, table_partitions, handle, true))) {
          LOG_WARN("fail to put and get partitions info to cache", K(ret), K(key));
        }
      }

      // unlock
      if (fill_cache_locks_[bucket_id].self_locked()) {
        fill_cache_locks_[bucket_id].unlock();
      }
    }
  } else {
    ObExternalTablePartitions pts;
    if (OB_FAIL(get_partitions_info(table_schema, sql_schema_guard, allocator, pts))) {
      LOG_WARN("fail to get partitions info", K(ret), K(table_schema));
    }
    table_partitions = &pts;
  }

  if (OB_SUCC(ret) && OB_NOT_NULL(table_partitions)) {
    const common::ObArrayWrap<PartitionInfo> &infos = table_partitions->partition_infos_;
    for (int i = 0; OB_SUCC(ret) && i < infos.count(); ++i) {
      const PartitionInfo &cache_partition_info = infos.at(i);

      PartitionInfo *out_partition_info;
      if (OB_ISNULL(out_partition_info = OB_NEWx(PartitionInfo, &allocator))) {
        ret = OB_ALLOCATE_MEMORY_FAILED;
        LOG_WARN("failed to allocate place holder for partition info", K(ret));
      } else {
        out_partition_info->modify_ts_ = cache_partition_info.modify_ts_;
      }
      OZ(ob_write_string(allocator,
                         cache_partition_info.partition_,
                         out_partition_info->partition_));
      OZ(ob_write_string(allocator, cache_partition_info.path_, out_partition_info->path_));
      if (OB_SUCC(ret) && cache_partition_info.partition_values_.count() > 0) {
        if (OB_FAIL(out_partition_info->partition_values_.allocate_array(
                allocator,
                cache_partition_info.partition_values_.count()))) {
          LOG_WARN("fail to allocate array for partition values", KR(ret));
        }
        for (int64_t j = 0; OB_SUCC(ret) && j < cache_partition_info.partition_values_.count();
             ++j) {
          OZ(ob_write_string(allocator,
                             cache_partition_info.partition_values_.at(j),
                             out_partition_info->partition_values_.at(j)));
        }
      }
      OZ(partition_infos.push_back(out_partition_info));
    }
  }
  return ret;
}

int ObExternalTableFileManager::get_partitions_info(const ObTableSchema &table_schema,
                                                    ObSqlSchemaGuard &sql_schema_guard,
                                                    common::ObIAllocator &allocator,
                                                    ObExternalTablePartitions &pts)
{
  int ret = OB_SUCCESS;
  const share::ObILakeTableMetadata *metadata
      = sql_schema_guard.get_table_metadata(table_schema.get_table_id());
  ObArenaAllocator tmp_allocator;
  ObCachedCatalogMetaGetter catalog_meta_getter{*sql_schema_guard.get_schema_guard(),
                                                tmp_allocator};
  Partitions partitions;
  if (OB_ISNULL(metadata)) {
    ret = OB_ERR_UNEXPECTED;
    LOG_WARN("get unexpected null metadata", K(ret), K(table_schema));
  } else {
    pts.create_ts_ = ObTimeUtil::current_time_ms();
    if (table_schema.get_partition_key_column_num() == 0) {
      // 非分区表
      OZ(pts.partition_infos_.allocate_array(allocator, 1));
      OZ(ob_write_string(allocator,
                         table_schema.get_external_file_location(),
                         pts.partition_infos_.at(0).path_));
      if (OB_SUCC(ret)) {
        pts.partition_infos_.at(0).partition_values_.reset();
        pts.partition_infos_.at(0).modify_ts_ = metadata->lake_table_metadata_version_;
      }
    } else {
      // 分区表
      if (OB_FAIL(catalog_meta_getter.fetch_partitions(allocator, metadata, partitions))) {
        LOG_WARN("failed to fetch partitions", K(ret), K(metadata));
      } else if (partitions.size() == 0) {
        // empty table, do nothing
      } else if (OB_FAIL(pts.partition_infos_.allocate_array(allocator, partitions.size()))) {
        LOG_WARN("fail to new partition info", KR(ret));
      } else {
        for (int64_t i = 0; OB_SUCC(ret) && i < partitions.size(); ++i) {
          const Apache::Hadoop::Hive::Partition &hive_part = partitions.at(i);
          PartitionInfo &p_info = pts.partition_infos_.at(i);
          if (OB_FAIL(
                  p_info.partition_values_.allocate_array(allocator, hive_part.values.size()))) {
            LOG_WARN("fail to new partition values", KR(ret));
          }
          for (int64_t j = 0; OB_SUCC(ret) && j < hive_part.values.size(); ++j) {
            ObString partition_value(hive_part.values[j].c_str());
            OZ(ob_write_string(allocator, partition_value, p_info.partition_values_.at(j)));
          }
          ObString path(hive_part.sd.location.c_str());
          OZ(ob_write_string(allocator, path, p_info.path_));
          std::map<std::string, std::string>::const_iterator params_iter;
          if (OB_SUCC(ret)) {
            params_iter = hive_part.parameters.find(share::LAST_DDL_TIME);
            if (OB_UNLIKELY(params_iter != hive_part.parameters.end())) {
              p_info.modify_ts_ = ::obsys::ObStringUtil::str_to_int(params_iter->second.c_str(), 0);
              LOG_TRACE("get latest ddl time", K(ret), K(p_info.modify_ts_));
            } else {
              p_info.modify_ts_ = 0;
              LOG_TRACE("not contain latest ddl time", K(ret));
            }
          }
        }
      }
    }
  }

  return ret;
}

int ObExternalTableFileManager::update_inner_table_files_list_by_part(
    ObMySQLTransaction &trans,
    const uint64_t tenant_id,
    const uint64_t table_id,
    const uint64_t partition_id,
    const ObIArray<ObExternalFileInfoTmp> &file_infos,
    ObIArray<uint64_t> &updated_part_ids)
{
  int ret = OB_SUCCESS;
  int64_t cur_time = ObTimeUtil::current_time();
  ObSEArray<ObExternalFileInfoTmp, 16> old_file_infos;
  ObSEArray<int64_t, 16> old_file_ids;
  ObSEArray<ObExternalFileInfoTmp, 16> insert_file_infos;
  ObSEArray<int64_t, 16> insert_file_ids;
  ObSEArray<ObExternalFileInfoTmp, 16> update_file_infos;
  ObSEArray<int64_t, 16> update_file_ids;
  ObSEArray<ObExternalFileInfoTmp, 16> delete_file_infos;
  ObSEArray<int64_t, 16> delete_file_ids;
  bool updated = false;
  ObArenaAllocator allocator;
  ObSqlString update_sql;
  ObSqlString insert_sql;
  ObSqlString delete_sql;
  int64_t update_rows = 0;
  int64_t insert_rows = 0;
  int64_t max_file_id = 0;// ObCSVTableRowIterator::MIN_EXTERNAL_TABLE_FILE_ID - 1
  common::hash::ObHashMap<ObString, int64_t> hash_map;
  char file_url_buf[256] = { 0 };
  char file_content_digest_buf[128] = { 0 };
  bool is_odps_external_table = false;
  uint64_t data_version = OB_INVALID_VERSION;
  if (OB_FAIL(GET_MIN_DATA_VERSION(tenant_id, data_version))) {
    LOG_WARN("get min data_version failed", KR(ret), K(tenant_id));
  } else if (OB_FAIL(ObSQLUtils::is_odps_external_table(tenant_id, table_id, is_odps_external_table))) {
    LOG_WARN("failed to check is odps external table or not", K(ret), K(tenant_id), K(table_id));
  }
  OZ(get_all_records_from_inner_table(allocator, tenant_id, table_id, partition_id, old_file_infos, old_file_ids));
  OZ(hash_map.create(std::max(file_infos.count(), old_file_infos.count()) + 1, "ExternalFile"));
  for (int64_t i = 0; OB_SUCC(ret) && i < old_file_infos.count(); i++) {
    OZ(hash_map.set_refactored(old_file_infos.at(i).file_url_, i));
    max_file_id = old_file_ids.at(i) > max_file_id ? old_file_ids.at(i) : max_file_id;
  }

  for (int64_t i = 0; OB_SUCC(ret) && i < file_infos.count(); i++) {
    int64_t file_idx = 0;
    ret = hash_map.get_refactored(file_infos.at(i).file_url_, file_idx);
    if (ret == OB_HASH_NOT_EXIST) {
      ret = OB_SUCCESS;
      OZ(insert_file_infos.push_back(file_infos.at(i)));
      OZ(insert_file_ids.push_back(is_odps_external_table ? 0 : ++max_file_id)); // odps table's file_id is 0
    } else if (ret == OB_SUCCESS) {
      if (old_file_infos.at(file_idx).file_size_ != file_infos.at(i).file_size_
          || old_file_infos.at(file_idx).content_digest_ != file_infos.at(i).content_digest_
          || old_file_infos.at(file_idx).modify_time_ != file_infos.at(i).modify_time_
          || old_file_infos.at(file_idx).delete_version_ != MAX_VERSION) {
        OZ(update_file_infos.push_back(file_infos.at(i)));
        OZ(update_file_ids.push_back(old_file_ids.at(file_idx)));
      }
    } else {
      LOG_WARN("unexpected error", K(ret), K(i));
    }
  }
  OZ(hash_map.reuse());
  for (int64_t i = 0; OB_SUCC(ret) && i < file_infos.count(); i++) {
    OZ(hash_map.set_refactored(file_infos.at(i).file_url_, i)); // odps table's file_id is 0
  }
  for (int64_t i = 0; OB_SUCC(ret) && i < old_file_infos.count(); i++) {
    int64_t existed = 0;
    OZ(hash_map.get_refactored(old_file_infos.at(i).file_url_, existed));
    if (ret == OB_HASH_NOT_EXIST) {
      ret = OB_SUCCESS;
      OZ(delete_file_infos.push_back(old_file_infos.at(i)));
      OZ(delete_file_ids.push_back(old_file_ids.at(i)));
    }
  }
  if (OB_SUCC(ret) && delete_file_infos.count() > 0) {
    OZ(delete_sql.assign_fmt("UPDATE %s SET DELETE_VERSION = %ld WHERE (TABLE_ID, PART_ID, FILE_ID) IN (",
                              OB_ALL_EXTERNAL_TABLE_FILE_TNAME, cur_time));
    for (int64_t i = 0; OB_SUCC(ret) && i < delete_file_infos.count(); i++) {
      OZ(delete_sql.append_fmt("%c(%ld, %ld, %ld)", (0 == i) ? ' ' : ',', table_id, partition_id,
                                                      delete_file_ids.at(i)));
    }
    OZ(delete_sql.append(")"));
    OZ(trans.write(tenant_id, delete_sql.ptr(), update_rows));
    updated = true;
  }
  if (OB_SUCC(ret) && update_file_infos.count() > 0) {
    for (int64_t i = 0; OB_SUCC(ret) && i < update_file_infos.count(); i++) {
      if (data_version >= DATA_VERSION_4_4_1_0) {
        ObString content_digest = update_file_infos.at(i).content_digest_;
        int new_digest_len = ObHexEscapeSqlStr(update_file_infos.at(i).content_digest_)
                               .to_string(file_content_digest_buf, 128);
        content_digest.assign(file_content_digest_buf, new_digest_len);
        OZ(update_sql.assign_fmt(
          "UPDATE %s SET"
          " CREATE_VERSION = CASE WHEN DELETE_VERSION != %ld THEN %ld ELSE CREATE_VERSION end,"
          " DELETE_VERSION = %ld, FILE_SIZE = %ld, LAST_MODIFY_TIME = %ld, CONTENT_DIGEST = '%.*s'"
          " WHERE TABLE_ID = %lu AND PART_ID = %lu AND FILE_ID=%ld",
          OB_ALL_EXTERNAL_TABLE_FILE_TNAME, MAX_VERSION, cur_time, MAX_VERSION,
          update_file_infos.at(i).file_size_, update_file_infos.at(i).modify_time_,
          content_digest.length(), content_digest.ptr(), table_id, partition_id,
          update_file_ids.at(i)));
      } else {
        OZ(update_sql.assign_fmt(
          "UPDATE %s SET"
          " CREATE_VERSION = CASE WHEN DELETE_VERSION != %ld THEN %ld ELSE CREATE_VERSION end,"
          " DELETE_VERSION = %ld, FILE_SIZE = %ld WHERE TABLE_ID = %lu AND PART_ID = %lu AND FILE_ID=%ld",
          OB_ALL_EXTERNAL_TABLE_FILE_TNAME, MAX_VERSION, cur_time, MAX_VERSION,
          update_file_infos.at(i).file_size_, table_id, partition_id, update_file_ids.at(i)));
      }
      OZ(trans.write(tenant_id, update_sql.ptr(), update_rows));
    }
    updated = true;
  }
  if (OB_SUCC(ret) && insert_file_infos.count() > 0) {
    if (data_version >= DATA_VERSION_4_4_1_0) {
      OZ(insert_sql.assign_fmt(
        "INSERT INTO %s(TABLE_ID,PART_ID,FILE_ID,FILE_URL,CONTENT_DIGEST,CREATE_VERSION,"
        "DELETE_VERSION,FILE_SIZE,LAST_MODIFY_TIME) VALUES",
        OB_ALL_EXTERNAL_TABLE_FILE_TNAME));
      for (int64_t i = 0; OB_SUCC(ret) && i < insert_file_infos.count(); i++) {
        ObString file_url = insert_file_infos.at(i).file_url_;
        int new_url_len =
          ObHexEscapeSqlStr(insert_file_infos.at(i).file_url_).to_string(file_url_buf, 256);
        file_url.assign(file_url_buf, new_url_len);
        ObString content_digest = insert_file_infos.at(i).content_digest_;
        int new_digest_len = ObHexEscapeSqlStr(insert_file_infos.at(i).content_digest_)
                               .to_string(file_content_digest_buf, 128);
        content_digest.assign(file_content_digest_buf, new_digest_len);
        OZ(insert_sql.append_fmt(
          "%c(%lu,%lu,%ld,'%.*s','%.*s',%ld,%ld,%ld,%ld)", (0 == i) ? ' ' : ',', table_id,
          partition_id, insert_file_ids.at(i), file_url.length(), file_url.ptr(),
          content_digest.length(), content_digest.ptr(), cur_time, MAX_VERSION,
          insert_file_infos.at(i).file_size_, insert_file_infos.at(i).modify_time_));
      }
    } else {
      OZ(insert_sql.assign_fmt(
        "INSERT INTO %s(TABLE_ID,PART_ID,FILE_ID,FILE_URL,CREATE_VERSION,DELETE_VERSION,FILE_SIZE) VALUES",
        OB_ALL_EXTERNAL_TABLE_FILE_TNAME));
      for (int64_t i = 0; OB_SUCC(ret) && i < insert_file_infos.count(); i++) {
        ObString file_url = insert_file_infos.at(i).file_url_;
        int new_url_len =
          ObHexEscapeSqlStr(insert_file_infos.at(i).file_url_).to_string(file_url_buf, 256);
        file_url.assign(file_url_buf, new_url_len);
        OZ(insert_sql.append_fmt("%c(%lu,%lu,%ld,'%.*s',%ld,%ld,%ld)", (0 == i) ? ' ' : ',',
                                 table_id, partition_id, insert_file_ids.at(i), file_url.length(),
                                 file_url.ptr(), cur_time, MAX_VERSION,
                                 insert_file_infos.at(i).file_size_));
      }
    }
    OZ(trans.write(tenant_id, insert_sql.ptr(), insert_rows));
    updated = true;
  }
  if (OB_SUCC(ret) && updated) {
    OZ (updated_part_ids.push_back(partition_id));
  }

  return ret;
}

int ObExternalTableFileManager::get_all_records_from_inner_table(ObIAllocator &allocator,
                                                                  int64_t tenant_id,
                                                                  int64_t table_id,
                                                                  int64_t partition_id,
                                                                  ObIArray<ObExternalFileInfoTmp> &file_urls,
                                                                  ObIArray<int64_t> &file_ids)
{
  int ret = OB_SUCCESS;
  uint64_t data_version = OB_INVALID_VERSION;
  if (OB_FAIL(GET_MIN_DATA_VERSION(tenant_id, data_version))) {
    LOG_WARN("get min data_version failed", KR(ret), K(tenant_id));
  } else {
    SMART_VAR(ObMySQLProxy::MySQLResult, res) {
      sqlclient::ObMySQLResult *result = NULL;
      ObSqlString sql;
      if (data_version >= DATA_VERSION_4_4_1_0) {
        OZ(sql.append_fmt("SELECT file_url, content_digest, file_id, file_size, last_modify_time,"
                          " delete_version FROM %s WHERE table_id = %lu AND part_id = %lu",
                          OB_ALL_EXTERNAL_TABLE_FILE_TNAME, table_id, partition_id));
      } else {
        OZ(sql.append_fmt("SELECT file_url, file_id, file_size, delete_version FROM %s"
                          " WHERE table_id = %lu AND part_id = %lu",
                          OB_ALL_EXTERNAL_TABLE_FILE_TNAME, table_id, partition_id));
      }
      OZ (GCTX.sql_proxy_->read(res, tenant_id, sql.ptr()));
      if (OB_SUCC(ret)) {
        if (OB_ISNULL(result = res.get_result())) {
          ret = OB_ERR_UNEXPECTED;
          LOG_WARN("result is null", K(ret));
        } else {
          while (OB_SUCC(ret) && OB_SUCC(result->next())) {
            ObString file_url;
            ObString content_digest;
            ObString tmp_content_digest;
            int64_t file_id = 0;
            int64_t file_size = 0;
            int64_t modify_time = 0;
            int64_t delete_version = 0;
            EXTRACT_VARCHAR_FIELD_MYSQL(*result, "file_url", file_url);
            EXTRACT_INT_FIELD_MYSQL(*result, "file_id", file_id, int64_t);
            EXTRACT_INT_FIELD_MYSQL(*result, "file_size", file_size, int64_t);
            EXTRACT_INT_FIELD_MYSQL(*result, "delete_version", delete_version, int64_t);
            ObString tmp_url;
            OZ (ob_write_string(allocator, file_url, tmp_url));
            if (data_version >= DATA_VERSION_4_4_1_0) {
              EXTRACT_INT_FIELD_MYSQL(*result, "last_modify_time", modify_time, int64_t);
              EXTRACT_VARCHAR_FIELD_MYSQL(*result, "content_digest", content_digest);
              OZ (ob_write_string(allocator, content_digest, tmp_content_digest));
            }
            ObExternalFileInfoTmp file_info;
            file_info.part_id_ = partition_id;
            file_info.file_url_ = tmp_url;
            file_info.content_digest_ = tmp_content_digest;
            file_info.file_size_ = file_size;
            file_info.modify_time_ = modify_time;
            file_info.delete_version_ = delete_version;
            OZ (file_urls.push_back(file_info));
            OZ (file_ids.push_back(file_id));
          }
          if (OB_FAIL(ret) && OB_ITER_END != ret) {
            LOG_WARN("get next result failed", K(ret));
          } else {
            ret = OB_SUCCESS;
          }
        }
      }
    }
  }
  return ret;
}
int ObExternalTableFileManager::fill_cache_from_inner_table(
    const ObExternalTableFilesKey &key,
    const ObExternalTableFiles *&ext_files,
    ObKVCacheHandle &handle)
{
  int ret = OB_SUCCESS;

  //only one worker need do the job
  int64_t bucket_id = key.hash() % LOAD_CACHE_LOCK_CNT;
  int64_t total_wait_secs = 0;
  uint64_t data_version = OB_INVALID_VERSION;

  while (OB_FAIL(fill_cache_locks_[bucket_id].lock(LOCK_TIMEOUT))
         && OB_TIMEOUT == ret && OB_SUCC(THIS_WORKER.check_status())) {
    total_wait_secs += (LOCK_TIMEOUT / 1000000);
    LOG_WARN("fill external table cache wait", K(total_wait_secs));
  }
  if (OB_SUCC(ret)) {
    //try fetch again
    if (OB_FAIL(GET_MIN_DATA_VERSION(MTL_ID(), data_version))) {
      LOG_WARN("get min data_version failed", KR(ret), K(MTL_ID()));
    } else if (OB_FAIL(kv_cache_.get(key, ext_files, handle))) {
      if (OB_ENTRY_NOT_EXIST != ret) {
        LOG_WARN("fail to get from KVCache", K(ret), K(key));
      }
    }

    if ((OB_SUCC(ret) && is_cache_value_timeout(*ext_files))
        || OB_ENTRY_NOT_EXIST == ret) {
      ret = OB_SUCCESS;
      SMART_VAR(ObMySQLProxy::MySQLResult, res) {
        sqlclient::ObMySQLResult *result = NULL;
        ObSqlString sql;
        int64_t cur_time = ObTimeUtil::current_time();

        if (OB_ISNULL(GCTX.sql_proxy_)) {
          ret = OB_ERR_UNEXPECTED;
        }

        if (data_version >= DATA_VERSION_4_4_1_0) {
          OZ(sql.append_fmt("SELECT file_url, content_digest, file_id, file_size, last_modify_time"
                            " FROM %s WHERE table_id = %lu AND part_id = %lu"
                            " AND create_version <=%ld AND %ld < delete_version",
                            OB_ALL_EXTERNAL_TABLE_FILE_TNAME, key.table_id_, key.partition_id_,
                            cur_time, cur_time));
        } else {
          OZ(sql.append_fmt("SELECT file_url, file_id, file_size FROM %s"
                            " WHERE table_id = %lu AND part_id = %lu"
                            " AND create_version <=%ld AND %ld < delete_version",
                            OB_ALL_EXTERNAL_TABLE_FILE_TNAME, key.table_id_, key.partition_id_,
                            cur_time, cur_time));
        }
        OZ (GCTX.sql_proxy_->read(res, key.tenant_id_, sql.ptr()));

        if (OB_SUCC(ret)) {
          if (OB_ISNULL(result = res.get_result())) {
            ret = OB_ERR_UNEXPECTED;
            LOG_WARN("result is null", K(ret));
          } else {
            ObSEArray<ObString, 16> temp_file_urls;
            ObSEArray<int64_t, 16> temp_file_ids;
            ObSEArray<int64_t, 16> temp_file_sizes;
            ObSEArray<int64_t, 16> temp_modify_times;
            ObSEArray<ObString, 16> temp_content_digests;
            ObArenaAllocator allocator;
            while (OB_SUCC(ret) && OB_SUCC(result->next())) {
              ObString file_url;
              ObString tmp_url;
              ObString content_digest;
              ObString tmp_content_digest;
              int64_t file_id = INT64_MAX;
              int64_t file_size = 0;
              int64_t modify_time = -1;
              EXTRACT_VARCHAR_FIELD_MYSQL(*result, "file_url", tmp_url);
              EXTRACT_INT_FIELD_MYSQL(*result, "file_id", file_id, int64_t);
              EXTRACT_INT_FIELD_MYSQL(*result, "file_size", file_size, int64_t);
              OZ (ob_write_string(allocator, tmp_url, file_url));
              OZ (temp_file_urls.push_back(file_url));
              OZ (temp_file_ids.push_back(file_id));
              OZ (temp_file_sizes.push_back(file_size));
              if (data_version >= DATA_VERSION_4_4_1_0) {
                EXTRACT_INT_FIELD_MYSQL(*result, "last_modify_time", modify_time, int64_t);
                EXTRACT_VARCHAR_FIELD_MYSQL(*result, "content_digest", tmp_content_digest);
                OZ (ob_write_string(allocator, tmp_content_digest, content_digest));
              }
              OZ (temp_modify_times.push_back(modify_time));
              OZ (temp_content_digests.push_back(content_digest));
            }
            if (OB_FAIL(ret) && OB_ITER_END != ret) {
              LOG_WARN("get next result failed", K(ret));
            } else {
              ret = OB_SUCCESS;
            }
            if (OB_SUCC(ret)) {
              ObExternalTableFiles temp_ext_files;
              temp_ext_files.create_ts_ = cur_time;
              temp_ext_files.file_urls_ = ObArrayWrap<ObString>(temp_file_urls.get_data(), temp_file_urls.count());
              temp_ext_files.file_ids_ = ObArrayWrap<int64_t>(temp_file_ids.get_data(), temp_file_ids.count());
              temp_ext_files.file_sizes_ = ObArrayWrap<int64_t>(temp_file_sizes.get_data(), temp_file_sizes.count());
              temp_ext_files.modify_times_ = ObArrayWrap<int64_t>(temp_modify_times.get_data(), temp_modify_times.count());
              temp_ext_files.content_digests_ = ObArrayWrap<ObString>(temp_content_digests.get_data(), temp_content_digests.count());
              OZ (kv_cache_.put_and_fetch(key, temp_ext_files, ext_files, handle, true));
            }
            LOG_TRACE("external table file urls", K(temp_file_urls), K(key));
          }
        }
      }
      LOG_TRACE("external table fill cache", K(ext_files), K(key));
    }
  }
  if (fill_cache_locks_[bucket_id].self_locked()) {
    fill_cache_locks_[bucket_id].unlock();
  }
  return ret;
}

int ObExternalTableFileManager::lock_for_refresh(
    ObMySQLTransaction &trans,
    const uint64_t tenant_id,
    const uint64_t object_id)
{
  int ret = OB_SUCCESS;
  ObInnerSQLConnection *conn = NULL;
  if (OB_ISNULL(conn = dynamic_cast<ObInnerSQLConnection *>(trans.get_connection()))) {
    ret = OB_ERR_UNEXPECTED;
    LOG_WARN("conn_ is NULL", KR(ret));
  } else {
    ObLockObjRequest lock_arg;
    lock_arg.obj_type_ = ObLockOBJType::OBJ_TYPE_EXTERNAL_TABLE_REFRESH;
    lock_arg.obj_id_ = object_id;
    lock_arg.lock_mode_ = EXCLUSIVE;
    lock_arg.op_type_ = ObTableLockOpType::IN_TRANS_COMMON_LOCK;
    lock_arg.timeout_us_ = 1000L * 1000L * 2; //2s
    lock_arg.is_from_sql_ = true;
    if (OB_FAIL(lock_arg.owner_id_.convert_from_value(ObLockOwnerType::DEFAULT_OWNER_TYPE,
                                                      get_tid_cache()))) {
      LOG_WARN("failed to get owner id", K(ret), K(get_tid_cache()));
    } else {
      while (OB_SUCC(ret) && OB_FAIL(ObInnerConnectionLockUtil::lock_obj(tenant_id, lock_arg, conn))) {
        LOG_WARN("external table add object lock failed", K(ret));
        if (OB_ERR_EXCLUSIVE_LOCK_CONFLICT == ret) {
          // overwrite ret
          if (OB_FAIL(THIS_WORKER.check_status())) {
            LOG_WARN("worker status error", K(ret));
          }
        }
      }
    }
  }


  return ret;
}

int ObExternalTableFileManager::flush_external_file_cache(
    const uint64_t tenant_id,
    const uint64_t table_id,
    const uint64_t part_id,
    const ObIArray<ObAddr> &all_servers)
{
  int ret = OB_SUCCESS;
  ObArenaAllocator allocator;
  ObAsyncRpcTaskWaitContext<ObRpcAsyncFlushExternalTableKVCacheCallBack> context;
  int64_t send_task_count = 0;
  OZ (context.init());
  OZ (context.get_cb_list().reserve(all_servers.count()));
  for (int64_t i = 0; OB_SUCC(ret) && i < all_servers.count(); i++) {
    ObFlushExternalTableFileCacheReq req;
    int64_t timeout = ObExternalTableFileManager::CACHE_EXPIRE_TIME;
    req.tenant_id_ = tenant_id;
    req.table_id_ = table_id;
    req.partition_id_ = part_id;
    ObRpcAsyncFlushExternalTableKVCacheCallBack* async_cb = nullptr;
    if (OB_ISNULL(async_cb = OB_NEWx(ObRpcAsyncFlushExternalTableKVCacheCallBack, (&allocator), (&context)))) {
      ret = OB_ALLOCATE_MEMORY_FAILED;
      LOG_WARN("failed to allocate async cb memory", K(ret));
    }
    OZ (context.get_cb_list().push_back(async_cb));
    OZ (GCTX.external_table_proxy_->to(all_servers.at(i))
                                            .by(tenant_id)
                                            .timeout(timeout)
                                            .flush_file_kvcahce(req, async_cb));
    if (OB_SUCC(ret)) {
      send_task_count++;
    }
  }

  context.set_task_count(send_task_count);

  do {
    int temp_ret = context.wait_executing_tasks();
    if (OB_SUCCESS != temp_ret) {
      LOG_WARN("fail to wait executing task", K(temp_ret));
      if (OB_SUCC(ret)) {
        ret = temp_ret;
      }
    }
  } while(0);

  for (int64_t i = 0; OB_SUCC(ret) && i < context.get_cb_list().count(); i++) {
    ret = context.get_cb_list().at(i)->get_task_resp().rcode_.rcode_;
    if (OB_FAIL(ret)) {
      if (OB_TIMEOUT == ret) {
        // flush timeout is OK, because the file cache has already expire
        ret = OB_SUCCESS;
      } else {
        LOG_WARN("async flush kvcache process failed", K(ret));
      }
    }
  }
  for (int64_t i = 0; i < context.get_cb_list().count(); i++) {
    context.get_cb_list().at(i)->~ObRpcAsyncFlushExternalTableKVCacheCallBack();
  }
  return ret;
}

int ObExternalTableFileManager::refresh_external_table(const uint64_t tenant_id,
                                                       const uint64_t table_id,
                                                       ObSchemaGetterGuard &schema_guard,
                                                       ObExecContext &exec_ctx,
                                                       bool &has_partition_changed) {
  int ret = OB_SUCCESS;
  ObArray<ObString> file_urls;
  ObArray<int64_t> file_sizes;
  ObExprRegexpSessionVariables regexp_vars;
  const ObTableSchema *table_schema = NULL;
  OZ (schema_guard.get_table_schema(tenant_id,
                                    table_id,
                                    table_schema));
  CK (table_schema != NULL);
  OZ (refresh_external_table(tenant_id, table_schema, schema_guard, exec_ctx, has_partition_changed));
  return ret;
}

int ObExternalTableFileManager::refresh_external_table(const uint64_t tenant_id,
                                                       const ObTableSchema *table_schema,
                                                       ObSchemaGetterGuard &schema_guard,
                                                       ObExecContext &exec_ctx,
                                                       bool &has_partition_changed) {
  int ret = OB_SUCCESS;
  ObArray<share::ObExternalTableBasicFileInfo> basic_file_infos;
  ObExprRegexpSessionVariables regexp_vars;
  ObString file_location;
  ObString access_info;
  CK (OB_NOT_NULL(table_schema));
  CK (exec_ctx.get_my_session() != NULL);
  OZ (ObExternalTableUtils::get_external_file_location(*table_schema, schema_guard, exec_ctx.get_allocator(), file_location));
  OZ (ObExternalTableUtils::get_external_file_location_access_info(*table_schema, schema_guard, access_info));
  if (OB_SUCC(ret) && ObSQLUtils::is_external_files_on_local_disk(file_location)) {
    OZ (ObSQLUtils::check_location_access_priv(file_location, exec_ctx.get_my_session()));
  }
  ObSqlString full_path;
  CK (GCTX.location_service_);
  OZ (exec_ctx.get_my_session()->get_regexp_session_vars(regexp_vars));
  OZ (ObExternalTableUtils::collect_external_file_list(
              exec_ctx.get_my_session(),
              tenant_id,
              table_schema->get_table_id(),
              file_location,
              access_info,
              table_schema->get_external_file_pattern(),
              table_schema->get_external_properties(),
              table_schema->is_partitioned_table(),
              regexp_vars, exec_ctx.get_allocator(),
              full_path,
              basic_file_infos));
  //TODO [External Table] opt performance
  ObSEArray<ObAddr, 8> all_servers;
  ObSEArray<uint64_t, 64> updated_part_ids;
  OZ (GCTX.location_service_->external_table_get(tenant_id, all_servers));
  OZ (ObExternalTableFileManager::get_instance().update_inner_table_file_list(exec_ctx, tenant_id,
                          table_schema->get_table_id(), basic_file_infos, updated_part_ids, has_partition_changed));
  if (OB_SUCC(ret) && updated_part_ids.count() > 0) {
    if (table_schema->is_partitioned_table()) {
      for (int64_t i = 0; OB_SUCC(ret) && i < updated_part_ids.count(); i++) {
        OZ (ObExternalTableFileManager::get_instance().flush_external_file_cache(tenant_id, table_schema->get_table_id(),
                      updated_part_ids.at(i), all_servers));
      }
    } else {
      OZ (ObExternalTableFileManager::get_instance().flush_external_file_cache(tenant_id, table_schema->get_table_id(),
        table_schema->get_table_id(),  all_servers));
    }
  }
  return ret;
}

int ObExternalTableFileManager::auto_refresh_external_table(ObExecContext &exec_ctx, const int64_t interval) {
  int ret = OB_SUCCESS;
  ObMySQLTransaction trans;
  CK (exec_ctx.get_my_session() != NULL);
  CK (exec_ctx.get_sql_ctx()->schema_guard_ != NULL);
  CK (OB_NOT_NULL(GCTX.sql_proxy_),
      OB_NOT_NULL(GCTX.schema_service_));
  uint64_t tenant_id = 0;
  if (OB_SUCC(ret)) {
    tenant_id = exec_ctx.get_my_session()->get_effective_tenant_id();
  }
  OZ (trans.start(GCTX.sql_proxy_, tenant_id));
  if (OB_SUCC(ret)) {
    if (interval == 0) {
      ObArray<const ObSimpleTableSchemaV2 *> table_schemas;
      OZ (exec_ctx.get_sql_ctx()->schema_guard_->get_table_schemas_in_tenant(tenant_id, table_schemas));
      for (int i = 0; OB_SUCC(ret) && i < table_schemas.count(); i++) {
        const ObSimpleTableSchemaV2 *simple_table = table_schemas.at(i);
        CK (simple_table != NULL);
        if (OB_SUCC(ret) && simple_table->get_table_type() == ObTableType::EXTERNAL_TABLE) {
          const ObTableSchema *table_schema = NULL;
          OZ (exec_ctx.get_sql_ctx()->schema_guard_->get_table_schema(tenant_id, simple_table->get_table_id(), table_schema));
          CK (table_schema != NULL);
          if (OB_SUCC(ret) && (2 == ((table_schema->get_table_flags() & 0B1100) >> 2))) {
            bool has_partition_changed = false;
            OZ (refresh_external_table(tenant_id, simple_table->get_table_id(), *exec_ctx.get_sql_ctx()->schema_guard_, exec_ctx, has_partition_changed));
          }
        }
      }
    } else if (interval == -1) {
      OZ (delete_auto_refresh_job(exec_ctx, trans));
    } else if (interval > 0) {
      OZ (delete_auto_refresh_job(exec_ctx, trans));
      OZ (create_auto_refresh_job(exec_ctx, interval, trans));
    } else {
      ret = OB_NOT_SUPPORTED;
      LOG_USER_ERROR(OB_NOT_SUPPORTED, "interval value");
      LOG_WARN("interval not supported", K(ret), K(interval));
    }

  }
  if (trans.is_started()) {
    int temp_ret = OB_SUCCESS;
    bool commit = OB_SUCC(ret);
    if (OB_SUCCESS != (temp_ret = trans.end(commit))) {
      ret = (OB_SUCC(ret)) ? temp_ret : ret;
      LOG_WARN("trans end failed", K(commit), K(temp_ret));
    }
  }
  return ret;
}


int ObExternalTableFileManager::delete_auto_refresh_job(ObExecContext &ctx, ObMySQLTransaction &trans) {
  int ret = OB_SUCCESS;
  ObSqlString sql;
  if (OB_FAIL(sql.append_fmt(
          "delete from %s where tenant_id = %lu and job_name= '%s'",
          share::OB_ALL_TENANT_SCHEDULER_JOB_TNAME,
          0UL, auto_refresh_job_name))) {
    LOG_WARN("failed to append fmt", K(ret));
  } else {
    int64_t affected_rows = 0;
    if (OB_ISNULL(ctx.get_my_session())) {
      ret = OB_ERR_UNEXPECTED;
      LOG_WARN("unexpected error", K(ret));
    } else if (OB_FAIL(trans.write(ctx.get_my_session()->get_effective_tenant_id(), sql.ptr(), affected_rows))) {
      LOG_WARN("execute sql failed", KR(ret), "sql", sql.ptr());
    }
  }
  return ret;
}

int ObExternalTableFileManager::create_auto_refresh_job(ObExecContext &ctx, const int64_t interval, ObMySQLTransaction &trans) {
  int ret = OB_SUCCESS;
  char buf[OB_MAX_PROC_ENV_LENGTH];
  int64_t pos = 0;
  CK (ctx.get_my_session() != NULL);
  OZ (sql::ObExecEnv::gen_exec_env(*ctx.get_my_session(), buf, OB_MAX_PROC_ENV_LENGTH, pos));
  ObString exec_env(pos, buf);
  ObCommonID raw_id;
  bool is_oracle_mode = false;
  OZ (ObCompatModeGetter::check_is_oracle_mode_with_tenant_id(ctx.get_my_session()->get_effective_tenant_id(), is_oracle_mode));
  OZ (storage::ObCommonIDUtils::gen_unique_id(ctx.get_my_session()->get_effective_tenant_id(), raw_id));
  int64_t max_job_id = raw_id.id() + dbms_scheduler::ObDBMSSchedTableOperator::JOB_ID_OFFSET;
  ObSqlString interval_str;
  int64_t tenant_id = ctx.get_my_session()->get_effective_tenant_id();
  OZ (interval_str.append_fmt("FREQ=SECONDLY; INTERVAL=%ld", interval));
  if (OB_SUCC(ret)) {
    HEAP_VAR(ObDBMSSchedJobInfo, job_info) {
      job_info.tenant_id_ = tenant_id;
      job_info.job_ = max_job_id;
      job_info.job_name_ = ObString(auto_refresh_job_name);
      job_info.job_action_ = ObString("dbms_external_table.auto_refresh_external_table()");
      job_info.lowner_ = is_oracle_mode ? "SYS" : "root@%";
      job_info.cowner_ = is_oracle_mode ? "SYS" : "oceanbase";
      job_info.powner_ = is_oracle_mode ? "SYS" : "root@%";
      job_info.start_date_ = ObTimeUtility::current_time();
      job_info.end_date_ = 64060560000000000;//4000-01-01 00:00:00.000000
      job_info.repeat_interval_ = interval_str.string();
      job_info.job_style_ = ObString("regular");
      job_info.job_type_ = ObString("STORED_PROCEDURE");
      job_info.job_class_ = ObString("DEFAULT_JOB_CLASS");
      job_info.enabled_ = true;
      job_info.auto_drop_ = false;
      job_info.max_run_duration_ = 24 * 60 * 60; //one day
      job_info.exec_env_ = exec_env;
      job_info.func_type_ = dbms_scheduler::ObDBMSSchedFuncType::EXT_FILE_REFRESH_JOB;
      if (OB_FAIL(ObDBMSSchedJobUtils::create_dbms_sched_job(
              trans, tenant_id, max_job_id, job_info))) {
        LOG_WARN("failed to create dbms scheduler job", KR(ret));
      }
    }
  }
  return ret;
}

}
}
